Statistics
| Branch: | Tag: | Revision:

root / lib / cmdlib.py @ e311ed53

History | View | Annotate | Download (340.7 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
import OpenSSL
37

    
38
from ganeti import ssh
39
from ganeti import utils
40
from ganeti import errors
41
from ganeti import hypervisor
42
from ganeti import locking
43
from ganeti import constants
44
from ganeti import objects
45
from ganeti import serializer
46
from ganeti import ssconf
47
from ganeti import uidpool
48
from ganeti import compat
49

    
50

    
51
class LogicalUnit(object):
52
  """Logical Unit base class.
53

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

63
  Note that all commands require root permissions.
64

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

68
  """
69
  HPATH = None
70
  HTYPE = None
71
  _OP_REQP = []
72
  REQ_BGL = True
73

    
74
  def __init__(self, processor, op, context, rpc):
75
    """Constructor for LogicalUnit.
76

77
    This needs to be overridden in derived classes in order to check op
78
    validity.
79

80
    """
81
    self.proc = processor
82
    self.op = op
83
    self.cfg = context.cfg
84
    self.context = context
85
    self.rpc = rpc
86
    # Dicts used to declare locking needs to mcpu
87
    self.needed_locks = None
88
    self.acquired_locks = {}
89
    self.share_locks = dict.fromkeys(locking.LEVELS, 0)
90
    self.add_locks = {}
91
    self.remove_locks = {}
92
    # Used to force good behavior when calling helper functions
93
    self.recalculate_locks = {}
94
    self.__ssh = None
95
    # logging
96
    self.LogWarning = processor.LogWarning # pylint: disable-msg=C0103
97
    self.LogInfo = processor.LogInfo # pylint: disable-msg=C0103
98
    self.LogStep = processor.LogStep # pylint: disable-msg=C0103
99
    # support for dry-run
100
    self.dry_run_result = None
101
    # support for generic debug attribute
102
    if (not hasattr(self.op, "debug_level") or
103
        not isinstance(self.op.debug_level, int)):
104
      self.op.debug_level = 0
105

    
106
    # Tasklets
107
    self.tasklets = None
108

    
109
    for attr_name in self._OP_REQP:
110
      attr_val = getattr(op, attr_name, None)
111
      if attr_val is None:
112
        raise errors.OpPrereqError("Required parameter '%s' missing" %
113
                                   attr_name, errors.ECODE_INVAL)
114

    
115
    self.CheckArguments()
116

    
117
  def __GetSSH(self):
118
    """Returns the SshRunner object
119

120
    """
121
    if not self.__ssh:
122
      self.__ssh = ssh.SshRunner(self.cfg.GetClusterName())
123
    return self.__ssh
124

    
125
  ssh = property(fget=__GetSSH)
126

    
127
  def CheckArguments(self):
128
    """Check syntactic validity for the opcode arguments.
129

130
    This method is for doing a simple syntactic check and ensure
131
    validity of opcode parameters, without any cluster-related
132
    checks. While the same can be accomplished in ExpandNames and/or
133
    CheckPrereq, doing these separate is better because:
134

135
      - ExpandNames is left as as purely a lock-related function
136
      - CheckPrereq is run after we have acquired locks (and possible
137
        waited for them)
138

139
    The function is allowed to change the self.op attribute so that
140
    later methods can no longer worry about missing parameters.
141

142
    """
143
    pass
144

    
145
  def ExpandNames(self):
146
    """Expand names for this LU.
147

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

153
    LUs which implement this method must also populate the self.needed_locks
154
    member, as a dict with lock levels as keys, and a list of needed lock names
155
    as values. Rules:
156

157
      - use an empty dict if you don't need any lock
158
      - if you don't need any lock at a particular level omit that level
159
      - don't put anything for the BGL level
160
      - if you want all locks at a level use locking.ALL_SET as a value
161

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

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

170
    Examples::
171

172
      # Acquire all nodes and one instance
173
      self.needed_locks = {
174
        locking.LEVEL_NODE: locking.ALL_SET,
175
        locking.LEVEL_INSTANCE: ['instance1.example.tld'],
176
      }
177
      # Acquire just two nodes
178
      self.needed_locks = {
179
        locking.LEVEL_NODE: ['node1.example.tld', 'node2.example.tld'],
180
      }
181
      # Acquire no locks
182
      self.needed_locks = {} # No, you can't leave it to the default value None
183

184
    """
185
    # The implementation of this method is mandatory only if the new LU is
186
    # concurrent, so that old LUs don't need to be changed all at the same
187
    # time.
188
    if self.REQ_BGL:
189
      self.needed_locks = {} # Exclusive LUs don't need locks.
190
    else:
191
      raise NotImplementedError
192

    
193
  def DeclareLocks(self, level):
194
    """Declare LU locking needs for a level
195

196
    While most LUs can just declare their locking needs at ExpandNames time,
197
    sometimes there's the need to calculate some locks after having acquired
198
    the ones before. This function is called just before acquiring locks at a
199
    particular level, but after acquiring the ones at lower levels, and permits
200
    such calculations. It can be used to modify self.needed_locks, and by
201
    default it does nothing.
202

203
    This function is only called if you have something already set in
204
    self.needed_locks for the level.
205

206
    @param level: Locking level which is going to be locked
207
    @type level: member of ganeti.locking.LEVELS
208

209
    """
210

    
211
  def CheckPrereq(self):
212
    """Check prerequisites for this LU.
213

214
    This method should check that the prerequisites for the execution
215
    of this LU are fulfilled. It can do internode communication, but
216
    it should be idempotent - no cluster or system changes are
217
    allowed.
218

219
    The method should raise errors.OpPrereqError in case something is
220
    not fulfilled. Its return value is ignored.
221

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

225
    """
226
    if self.tasklets is not None:
227
      for (idx, tl) in enumerate(self.tasklets):
228
        logging.debug("Checking prerequisites for tasklet %s/%s",
229
                      idx + 1, len(self.tasklets))
230
        tl.CheckPrereq()
231
    else:
232
      raise NotImplementedError
233

    
234
  def Exec(self, feedback_fn):
235
    """Execute the LU.
236

237
    This method should implement the actual work. It should raise
238
    errors.OpExecError for failures that are somewhat dealt with in
239
    code, or expected.
240

241
    """
242
    if self.tasklets is not None:
243
      for (idx, tl) in enumerate(self.tasklets):
244
        logging.debug("Executing tasklet %s/%s", idx + 1, len(self.tasklets))
245
        tl.Exec(feedback_fn)
246
    else:
247
      raise NotImplementedError
248

    
249
  def BuildHooksEnv(self):
250
    """Build hooks environment for this LU.
251

252
    This method should return a three-node tuple consisting of: a dict
253
    containing the environment that will be used for running the
254
    specific hook for this LU, a list of node names on which the hook
255
    should run before the execution, and a list of node names on which
256
    the hook should run after the execution.
257

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

263
    No nodes should be returned as an empty list (and not None).
264

265
    Note that if the HPATH for a LU class is None, this function will
266
    not be called.
267

268
    """
269
    raise NotImplementedError
270

    
271
  def HooksCallBack(self, phase, hook_results, feedback_fn, lu_result):
272
    """Notify the LU about the results of its hooks.
273

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

280
    @param phase: one of L{constants.HOOKS_PHASE_POST} or
281
        L{constants.HOOKS_PHASE_PRE}; it denotes the hooks phase
282
    @param hook_results: the results of the multi-node hooks rpc call
283
    @param feedback_fn: function used send feedback back to the caller
284
    @param lu_result: the previous Exec result this LU had, or None
285
        in the PRE phase
286
    @return: the new Exec result, based on the previous result
287
        and hook results
288

289
    """
290
    # API must be kept, thus we ignore the unused argument and could
291
    # be a function warnings
292
    # pylint: disable-msg=W0613,R0201
293
    return lu_result
294

    
295
  def _ExpandAndLockInstance(self):
296
    """Helper function to expand and lock an instance.
297

298
    Many LUs that work on an instance take its name in self.op.instance_name
299
    and need to expand it and then declare the expanded name for locking. This
300
    function does it, and then updates self.op.instance_name to the expanded
301
    name. It also initializes needed_locks as a dict, if this hasn't been done
302
    before.
303

304
    """
305
    if self.needed_locks is None:
306
      self.needed_locks = {}
307
    else:
308
      assert locking.LEVEL_INSTANCE not in self.needed_locks, \
309
        "_ExpandAndLockInstance called with instance-level locks set"
310
    self.op.instance_name = _ExpandInstanceName(self.cfg,
311
                                                self.op.instance_name)
312
    self.needed_locks[locking.LEVEL_INSTANCE] = self.op.instance_name
313

    
314
  def _LockInstancesNodes(self, primary_only=False):
315
    """Helper function to declare instances' nodes for locking.
316

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

322
    It should be called from DeclareLocks, and for safety only works if
323
    self.recalculate_locks[locking.LEVEL_NODE] is set.
324

325
    In the future it may grow parameters to just lock some instance's nodes, or
326
    to just lock primaries or secondary nodes, if needed.
327

328
    If should be called in DeclareLocks in a way similar to::
329

330
      if level == locking.LEVEL_NODE:
331
        self._LockInstancesNodes()
332

333
    @type primary_only: boolean
334
    @param primary_only: only lock primary nodes of locked instances
335

336
    """
337
    assert locking.LEVEL_NODE in self.recalculate_locks, \
338
      "_LockInstancesNodes helper function called with no nodes to recalculate"
339

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

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

    
352
    if self.recalculate_locks[locking.LEVEL_NODE] == constants.LOCKS_REPLACE:
353
      self.needed_locks[locking.LEVEL_NODE] = wanted_nodes
354
    elif self.recalculate_locks[locking.LEVEL_NODE] == constants.LOCKS_APPEND:
355
      self.needed_locks[locking.LEVEL_NODE].extend(wanted_nodes)
356

    
357
    del self.recalculate_locks[locking.LEVEL_NODE]
358

    
359

    
360
class NoHooksLU(LogicalUnit): # pylint: disable-msg=W0223
361
  """Simple LU which runs no hooks.
362

363
  This LU is intended as a parent for other LogicalUnits which will
364
  run no hooks, in order to reduce duplicate code.
365

366
  """
367
  HPATH = None
368
  HTYPE = None
369

    
370
  def BuildHooksEnv(self):
371
    """Empty BuildHooksEnv for NoHooksLu.
372

373
    This just raises an error.
374

375
    """
376
    assert False, "BuildHooksEnv called for NoHooksLUs"
377

    
378

    
379
class Tasklet:
380
  """Tasklet base class.
381

382
  Tasklets are subcomponents for LUs. LUs can consist entirely of tasklets or
383
  they can mix legacy code with tasklets. Locking needs to be done in the LU,
384
  tasklets know nothing about locks.
385

386
  Subclasses must follow these rules:
387
    - Implement CheckPrereq
388
    - Implement Exec
389

390
  """
391
  def __init__(self, lu):
392
    self.lu = lu
393

    
394
    # Shortcuts
395
    self.cfg = lu.cfg
396
    self.rpc = lu.rpc
397

    
398
  def CheckPrereq(self):
399
    """Check prerequisites for this tasklets.
400

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

405
    The method should raise errors.OpPrereqError in case something is not
406
    fulfilled. Its return value is ignored.
407

408
    This method should also update all parameters to their canonical form if it
409
    hasn't been done before.
410

411
    """
412
    raise NotImplementedError
413

    
414
  def Exec(self, feedback_fn):
415
    """Execute the tasklet.
416

417
    This method should implement the actual work. It should raise
418
    errors.OpExecError for failures that are somewhat dealt with in code, or
419
    expected.
420

421
    """
422
    raise NotImplementedError
423

    
424

    
425
def _GetWantedNodes(lu, nodes):
426
  """Returns list of checked and expanded node names.
427

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

436
  """
437
  if not isinstance(nodes, list):
438
    raise errors.OpPrereqError("Invalid argument type 'nodes'",
439
                               errors.ECODE_INVAL)
440

    
441
  if not nodes:
442
    raise errors.ProgrammerError("_GetWantedNodes should only be called with a"
443
      " non-empty list of nodes whose name is to be expanded.")
444

    
445
  wanted = [_ExpandNodeName(lu.cfg, name) for name in nodes]
446
  return utils.NiceSort(wanted)
447

    
448

    
449
def _GetWantedInstances(lu, instances):
450
  """Returns list of checked and expanded instance names.
451

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

461
  """
462
  if not isinstance(instances, list):
463
    raise errors.OpPrereqError("Invalid argument type 'instances'",
464
                               errors.ECODE_INVAL)
465

    
466
  if instances:
467
    wanted = [_ExpandInstanceName(lu.cfg, name) for name in instances]
468
  else:
469
    wanted = utils.NiceSort(lu.cfg.GetInstanceList())
470
  return wanted
471

    
472

    
473
def _CheckOutputFields(static, dynamic, selected):
474
  """Checks whether all selected fields are valid.
475

476
  @type static: L{utils.FieldSet}
477
  @param static: static fields set
478
  @type dynamic: L{utils.FieldSet}
479
  @param dynamic: dynamic fields set
480

481
  """
482
  f = utils.FieldSet()
483
  f.Extend(static)
484
  f.Extend(dynamic)
485

    
486
  delta = f.NonMatching(selected)
487
  if delta:
488
    raise errors.OpPrereqError("Unknown output fields selected: %s"
489
                               % ",".join(delta), errors.ECODE_INVAL)
490

    
491

    
492
def _CheckBooleanOpField(op, name):
493
  """Validates boolean opcode parameters.
494

495
  This will ensure that an opcode parameter is either a boolean value,
496
  or None (but that it always exists).
497

498
  """
499
  val = getattr(op, name, None)
500
  if not (val is None or isinstance(val, bool)):
501
    raise errors.OpPrereqError("Invalid boolean parameter '%s' (%s)" %
502
                               (name, str(val)), errors.ECODE_INVAL)
503
  setattr(op, name, val)
504

    
505

    
506
def _CheckGlobalHvParams(params):
507
  """Validates that given hypervisor params are not global ones.
508

509
  This will ensure that instances don't get customised versions of
510
  global params.
511

512
  """
513
  used_globals = constants.HVC_GLOBALS.intersection(params)
514
  if used_globals:
515
    msg = ("The following hypervisor parameters are global and cannot"
516
           " be customized at instance level, please modify them at"
517
           " cluster level: %s" % utils.CommaJoin(used_globals))
518
    raise errors.OpPrereqError(msg, errors.ECODE_INVAL)
519

    
520

    
521
def _CheckNodeOnline(lu, node):
522
  """Ensure that a given node is online.
523

524
  @param lu: the LU on behalf of which we make the check
525
  @param node: the node to check
526
  @raise errors.OpPrereqError: if the node is offline
527

528
  """
529
  if lu.cfg.GetNodeInfo(node).offline:
530
    raise errors.OpPrereqError("Can't use offline node %s" % node,
531
                               errors.ECODE_INVAL)
532

    
533

    
534
def _CheckNodeNotDrained(lu, node):
535
  """Ensure that a given node is not drained.
536

537
  @param lu: the LU on behalf of which we make the check
538
  @param node: the node to check
539
  @raise errors.OpPrereqError: if the node is drained
540

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

    
546

    
547
def _CheckNodeHasOS(lu, node, os_name, force_variant):
548
  """Ensure that a node supports a given OS.
549

550
  @param lu: the LU on behalf of which we make the check
551
  @param node: the node to check
552
  @param os_name: the OS to query about
553
  @param force_variant: whether to ignore variant errors
554
  @raise errors.OpPrereqError: if the node is not supporting the OS
555

556
  """
557
  result = lu.rpc.call_os_get(node, os_name)
558
  result.Raise("OS '%s' not in supported OS list for node %s" %
559
               (os_name, node),
560
               prereq=True, ecode=errors.ECODE_INVAL)
561
  if not force_variant:
562
    _CheckOSVariant(result.payload, os_name)
563

    
564

    
565
def _RequireFileStorage():
566
  """Checks that file storage is enabled.
567

568
  @raise errors.OpPrereqError: when file storage is disabled
569

570
  """
571
  if not constants.ENABLE_FILE_STORAGE:
572
    raise errors.OpPrereqError("File storage disabled at configure time",
573
                               errors.ECODE_INVAL)
574

    
575

    
576
def _CheckDiskTemplate(template):
577
  """Ensure a given disk template is valid.
578

579
  """
580
  if template not in constants.DISK_TEMPLATES:
581
    msg = ("Invalid disk template name '%s', valid templates are: %s" %
582
           (template, utils.CommaJoin(constants.DISK_TEMPLATES)))
583
    raise errors.OpPrereqError(msg, errors.ECODE_INVAL)
584
  if template == constants.DT_FILE:
585
    _RequireFileStorage()
586

    
587

    
588
def _CheckStorageType(storage_type):
589
  """Ensure a given storage type is valid.
590

591
  """
592
  if storage_type not in constants.VALID_STORAGE_TYPES:
593
    raise errors.OpPrereqError("Unknown storage type: %s" % storage_type,
594
                               errors.ECODE_INVAL)
595
  if storage_type == constants.ST_FILE:
596
    _RequireFileStorage()
597

    
598

    
599

    
600
def _CheckInstanceDown(lu, instance, reason):
601
  """Ensure that an instance is not running."""
602
  if instance.admin_up:
603
    raise errors.OpPrereqError("Instance %s is marked to be up, %s" %
604
                               (instance.name, reason), errors.ECODE_STATE)
605

    
606
  pnode = instance.primary_node
607
  ins_l = lu.rpc.call_instance_list([pnode], [instance.hypervisor])[pnode]
608
  ins_l.Raise("Can't contact node %s for instance information" % pnode,
609
              prereq=True, ecode=errors.ECODE_ENVIRON)
610

    
611
  if instance.name in ins_l.payload:
612
    raise errors.OpPrereqError("Instance %s is running, %s" %
613
                               (instance.name, reason), errors.ECODE_STATE)
614

    
615

    
616
def _ExpandItemName(fn, name, kind):
617
  """Expand an item name.
618

619
  @param fn: the function to use for expansion
620
  @param name: requested item name
621
  @param kind: text description ('Node' or 'Instance')
622
  @return: the resolved (full) name
623
  @raise errors.OpPrereqError: if the item is not found
624

625
  """
626
  full_name = fn(name)
627
  if full_name is None:
628
    raise errors.OpPrereqError("%s '%s' not known" % (kind, name),
629
                               errors.ECODE_NOENT)
630
  return full_name
631

    
632

    
633
def _ExpandNodeName(cfg, name):
634
  """Wrapper over L{_ExpandItemName} for nodes."""
635
  return _ExpandItemName(cfg.ExpandNodeName, name, "Node")
636

    
637

    
638
def _ExpandInstanceName(cfg, name):
639
  """Wrapper over L{_ExpandItemName} for instance."""
640
  return _ExpandItemName(cfg.ExpandInstanceName, name, "Instance")
641

    
642

    
643
def _BuildInstanceHookEnv(name, primary_node, secondary_nodes, os_type, status,
644
                          memory, vcpus, nics, disk_template, disks,
645
                          bep, hvp, hypervisor_name):
646
  """Builds instance related env variables for hooks
647

648
  This builds the hook environment from individual variables.
649

650
  @type name: string
651
  @param name: the name of the instance
652
  @type primary_node: string
653
  @param primary_node: the name of the instance's primary node
654
  @type secondary_nodes: list
655
  @param secondary_nodes: list of secondary nodes as strings
656
  @type os_type: string
657
  @param os_type: the name of the instance's OS
658
  @type status: boolean
659
  @param status: the should_run status of the instance
660
  @type memory: string
661
  @param memory: the memory size of the instance
662
  @type vcpus: string
663
  @param vcpus: the count of VCPUs the instance has
664
  @type nics: list
665
  @param nics: list of tuples (ip, mac, mode, link) representing
666
      the NICs the instance has
667
  @type disk_template: string
668
  @param disk_template: the disk template of the instance
669
  @type disks: list
670
  @param disks: the list of (size, mode) pairs
671
  @type bep: dict
672
  @param bep: the backend parameters for the instance
673
  @type hvp: dict
674
  @param hvp: the hypervisor parameters for the instance
675
  @type hypervisor_name: string
676
  @param hypervisor_name: the hypervisor for the instance
677
  @rtype: dict
678
  @return: the hook environment for this instance
679

680
  """
681
  if status:
682
    str_status = "up"
683
  else:
684
    str_status = "down"
685
  env = {
686
    "OP_TARGET": name,
687
    "INSTANCE_NAME": name,
688
    "INSTANCE_PRIMARY": primary_node,
689
    "INSTANCE_SECONDARIES": " ".join(secondary_nodes),
690
    "INSTANCE_OS_TYPE": os_type,
691
    "INSTANCE_STATUS": str_status,
692
    "INSTANCE_MEMORY": memory,
693
    "INSTANCE_VCPUS": vcpus,
694
    "INSTANCE_DISK_TEMPLATE": disk_template,
695
    "INSTANCE_HYPERVISOR": hypervisor_name,
696
  }
697

    
698
  if nics:
699
    nic_count = len(nics)
700
    for idx, (ip, mac, mode, link) in enumerate(nics):
701
      if ip is None:
702
        ip = ""
703
      env["INSTANCE_NIC%d_IP" % idx] = ip
704
      env["INSTANCE_NIC%d_MAC" % idx] = mac
705
      env["INSTANCE_NIC%d_MODE" % idx] = mode
706
      env["INSTANCE_NIC%d_LINK" % idx] = link
707
      if mode == constants.NIC_MODE_BRIDGED:
708
        env["INSTANCE_NIC%d_BRIDGE" % idx] = link
709
  else:
710
    nic_count = 0
711

    
712
  env["INSTANCE_NIC_COUNT"] = nic_count
713

    
714
  if disks:
715
    disk_count = len(disks)
716
    for idx, (size, mode) in enumerate(disks):
717
      env["INSTANCE_DISK%d_SIZE" % idx] = size
718
      env["INSTANCE_DISK%d_MODE" % idx] = mode
719
  else:
720
    disk_count = 0
721

    
722
  env["INSTANCE_DISK_COUNT"] = disk_count
723

    
724
  for source, kind in [(bep, "BE"), (hvp, "HV")]:
725
    for key, value in source.items():
726
      env["INSTANCE_%s_%s" % (kind, key)] = value
727

    
728
  return env
729

    
730

    
731
def _NICListToTuple(lu, nics):
732
  """Build a list of nic information tuples.
733

734
  This list is suitable to be passed to _BuildInstanceHookEnv or as a return
735
  value in LUQueryInstanceData.
736

737
  @type lu:  L{LogicalUnit}
738
  @param lu: the logical unit on whose behalf we execute
739
  @type nics: list of L{objects.NIC}
740
  @param nics: list of nics to convert to hooks tuples
741

742
  """
743
  hooks_nics = []
744
  c_nicparams = lu.cfg.GetClusterInfo().nicparams[constants.PP_DEFAULT]
745
  for nic in nics:
746
    ip = nic.ip
747
    mac = nic.mac
748
    filled_params = objects.FillDict(c_nicparams, nic.nicparams)
749
    mode = filled_params[constants.NIC_MODE]
750
    link = filled_params[constants.NIC_LINK]
751
    hooks_nics.append((ip, mac, mode, link))
752
  return hooks_nics
753

    
754

    
755
def _BuildInstanceHookEnvByObject(lu, instance, override=None):
756
  """Builds instance related env variables for hooks from an object.
757

758
  @type lu: L{LogicalUnit}
759
  @param lu: the logical unit on whose behalf we execute
760
  @type instance: L{objects.Instance}
761
  @param instance: the instance for which we should build the
762
      environment
763
  @type override: dict
764
  @param override: dictionary with key/values that will override
765
      our values
766
  @rtype: dict
767
  @return: the hook environment dictionary
768

769
  """
770
  cluster = lu.cfg.GetClusterInfo()
771
  bep = cluster.FillBE(instance)
772
  hvp = cluster.FillHV(instance)
773
  args = {
774
    'name': instance.name,
775
    'primary_node': instance.primary_node,
776
    'secondary_nodes': instance.secondary_nodes,
777
    'os_type': instance.os,
778
    'status': instance.admin_up,
779
    'memory': bep[constants.BE_MEMORY],
780
    'vcpus': bep[constants.BE_VCPUS],
781
    'nics': _NICListToTuple(lu, instance.nics),
782
    'disk_template': instance.disk_template,
783
    'disks': [(disk.size, disk.mode) for disk in instance.disks],
784
    'bep': bep,
785
    'hvp': hvp,
786
    'hypervisor_name': instance.hypervisor,
787
  }
788
  if override:
789
    args.update(override)
790
  return _BuildInstanceHookEnv(**args) # pylint: disable-msg=W0142
791

    
792

    
793
def _AdjustCandidatePool(lu, exceptions):
794
  """Adjust the candidate pool after node operations.
795

796
  """
797
  mod_list = lu.cfg.MaintainCandidatePool(exceptions)
798
  if mod_list:
799
    lu.LogInfo("Promoted nodes to master candidate role: %s",
800
               utils.CommaJoin(node.name for node in mod_list))
801
    for name in mod_list:
802
      lu.context.ReaddNode(name)
803
  mc_now, mc_max, _ = lu.cfg.GetMasterCandidateStats(exceptions)
804
  if mc_now > mc_max:
805
    lu.LogInfo("Note: more nodes are candidates (%d) than desired (%d)" %
806
               (mc_now, mc_max))
807

    
808

    
809
def _DecideSelfPromotion(lu, exceptions=None):
810
  """Decide whether I should promote myself as a master candidate.
811

812
  """
813
  cp_size = lu.cfg.GetClusterInfo().candidate_pool_size
814
  mc_now, mc_should, _ = lu.cfg.GetMasterCandidateStats(exceptions)
815
  # the new node will increase mc_max with one, so:
816
  mc_should = min(mc_should + 1, cp_size)
817
  return mc_now < mc_should
818

    
819

    
820
def _CheckNicsBridgesExist(lu, target_nics, target_node,
821
                               profile=constants.PP_DEFAULT):
822
  """Check that the brigdes needed by a list of nics exist.
823

824
  """
825
  c_nicparams = lu.cfg.GetClusterInfo().nicparams[profile]
826
  paramslist = [objects.FillDict(c_nicparams, nic.nicparams)
827
                for nic in target_nics]
828
  brlist = [params[constants.NIC_LINK] for params in paramslist
829
            if params[constants.NIC_MODE] == constants.NIC_MODE_BRIDGED]
830
  if brlist:
831
    result = lu.rpc.call_bridges_exist(target_node, brlist)
832
    result.Raise("Error checking bridges on destination node '%s'" %
833
                 target_node, prereq=True, ecode=errors.ECODE_ENVIRON)
834

    
835

    
836
def _CheckInstanceBridgesExist(lu, instance, node=None):
837
  """Check that the brigdes needed by an instance exist.
838

839
  """
840
  if node is None:
841
    node = instance.primary_node
842
  _CheckNicsBridgesExist(lu, instance.nics, node)
843

    
844

    
845
def _CheckOSVariant(os_obj, name):
846
  """Check whether an OS name conforms to the os variants specification.
847

848
  @type os_obj: L{objects.OS}
849
  @param os_obj: OS object to check
850
  @type name: string
851
  @param name: OS name passed by the user, to check for validity
852

853
  """
854
  if not os_obj.supported_variants:
855
    return
856
  try:
857
    variant = name.split("+", 1)[1]
858
  except IndexError:
859
    raise errors.OpPrereqError("OS name must include a variant",
860
                               errors.ECODE_INVAL)
861

    
862
  if variant not in os_obj.supported_variants:
863
    raise errors.OpPrereqError("Unsupported OS variant", errors.ECODE_INVAL)
864

    
865

    
866
def _GetNodeInstancesInner(cfg, fn):
867
  return [i for i in cfg.GetAllInstancesInfo().values() if fn(i)]
868

    
869

    
870
def _GetNodeInstances(cfg, node_name):
871
  """Returns a list of all primary and secondary instances on a node.
872

873
  """
874

    
875
  return _GetNodeInstancesInner(cfg, lambda inst: node_name in inst.all_nodes)
876

    
877

    
878
def _GetNodePrimaryInstances(cfg, node_name):
879
  """Returns primary instances on a node.
880

881
  """
882
  return _GetNodeInstancesInner(cfg,
883
                                lambda inst: node_name == inst.primary_node)
884

    
885

    
886
def _GetNodeSecondaryInstances(cfg, node_name):
887
  """Returns secondary instances on a node.
888

889
  """
890
  return _GetNodeInstancesInner(cfg,
891
                                lambda inst: node_name in inst.secondary_nodes)
892

    
893

    
894
def _GetStorageTypeArgs(cfg, storage_type):
895
  """Returns the arguments for a storage type.
896

897
  """
898
  # Special case for file storage
899
  if storage_type == constants.ST_FILE:
900
    # storage.FileStorage wants a list of storage directories
901
    return [[cfg.GetFileStorageDir()]]
902

    
903
  return []
904

    
905

    
906
def _FindFaultyInstanceDisks(cfg, rpc, instance, node_name, prereq):
907
  faulty = []
908

    
909
  for dev in instance.disks:
910
    cfg.SetDiskID(dev, node_name)
911

    
912
  result = rpc.call_blockdev_getmirrorstatus(node_name, instance.disks)
913
  result.Raise("Failed to get disk status from node %s" % node_name,
914
               prereq=prereq, ecode=errors.ECODE_ENVIRON)
915

    
916
  for idx, bdev_status in enumerate(result.payload):
917
    if bdev_status and bdev_status.ldisk_status == constants.LDS_FAULTY:
918
      faulty.append(idx)
919

    
920
  return faulty
921

    
922

    
923
def _FormatTimestamp(secs):
924
  """Formats a Unix timestamp with the local timezone.
925

926
  """
927
  return time.strftime("%F %T %Z", time.gmtime(secs))
928

    
929

    
930
class LUPostInitCluster(LogicalUnit):
931
  """Logical unit for running hooks after cluster initialization.
932

933
  """
934
  HPATH = "cluster-init"
935
  HTYPE = constants.HTYPE_CLUSTER
936
  _OP_REQP = []
937

    
938
  def BuildHooksEnv(self):
939
    """Build hooks env.
940

941
    """
942
    env = {"OP_TARGET": self.cfg.GetClusterName()}
943
    mn = self.cfg.GetMasterNode()
944
    return env, [], [mn]
945

    
946
  def CheckPrereq(self):
947
    """No prerequisites to check.
948

949
    """
950
    return True
951

    
952
  def Exec(self, feedback_fn):
953
    """Nothing to do.
954

955
    """
956
    return True
957

    
958

    
959
class LUDestroyCluster(LogicalUnit):
960
  """Logical unit for destroying the cluster.
961

962
  """
963
  HPATH = "cluster-destroy"
964
  HTYPE = constants.HTYPE_CLUSTER
965
  _OP_REQP = []
966

    
967
  def BuildHooksEnv(self):
968
    """Build hooks env.
969

970
    """
971
    env = {"OP_TARGET": self.cfg.GetClusterName()}
972
    return env, [], []
973

    
974
  def CheckPrereq(self):
975
    """Check prerequisites.
976

977
    This checks whether the cluster is empty.
978

979
    Any errors are signaled by raising errors.OpPrereqError.
980

981
    """
982
    master = self.cfg.GetMasterNode()
983

    
984
    nodelist = self.cfg.GetNodeList()
985
    if len(nodelist) != 1 or nodelist[0] != master:
986
      raise errors.OpPrereqError("There are still %d node(s) in"
987
                                 " this cluster." % (len(nodelist) - 1),
988
                                 errors.ECODE_INVAL)
989
    instancelist = self.cfg.GetInstanceList()
990
    if instancelist:
991
      raise errors.OpPrereqError("There are still %d instance(s) in"
992
                                 " this cluster." % len(instancelist),
993
                                 errors.ECODE_INVAL)
994

    
995
  def Exec(self, feedback_fn):
996
    """Destroys the cluster.
997

998
    """
999
    master = self.cfg.GetMasterNode()
1000
    modify_ssh_setup = self.cfg.GetClusterInfo().modify_ssh_setup
1001

    
1002
    # Run post hooks on master node before it's removed
1003
    hm = self.proc.hmclass(self.rpc.call_hooks_runner, self)
1004
    try:
1005
      hm.RunPhase(constants.HOOKS_PHASE_POST, [master])
1006
    except:
1007
      # pylint: disable-msg=W0702
1008
      self.LogWarning("Errors occurred running hooks on %s" % master)
1009

    
1010
    result = self.rpc.call_node_stop_master(master, False)
1011
    result.Raise("Could not disable the master role")
1012

    
1013
    if modify_ssh_setup:
1014
      priv_key, pub_key, _ = ssh.GetUserFiles(constants.GANETI_RUNAS)
1015
      utils.CreateBackup(priv_key)
1016
      utils.CreateBackup(pub_key)
1017

    
1018
    return master
1019

    
1020

    
1021
def _VerifyCertificateInner(filename, expired, not_before, not_after, now,
1022
                            warn_days=constants.SSL_CERT_EXPIRATION_WARN,
1023
                            error_days=constants.SSL_CERT_EXPIRATION_ERROR):
1024
  """Verifies certificate details for LUVerifyCluster.
1025

1026
  """
1027
  if expired:
1028
    msg = "Certificate %s is expired" % filename
1029

    
1030
    if not_before is not None and not_after is not None:
1031
      msg += (" (valid from %s to %s)" %
1032
              (_FormatTimestamp(not_before),
1033
               _FormatTimestamp(not_after)))
1034
    elif not_before is not None:
1035
      msg += " (valid from %s)" % _FormatTimestamp(not_before)
1036
    elif not_after is not None:
1037
      msg += " (valid until %s)" % _FormatTimestamp(not_after)
1038

    
1039
    return (LUVerifyCluster.ETYPE_ERROR, msg)
1040

    
1041
  elif not_before is not None and not_before > now:
1042
    return (LUVerifyCluster.ETYPE_WARNING,
1043
            "Certificate %s not yet valid (valid from %s)" %
1044
            (filename, _FormatTimestamp(not_before)))
1045

    
1046
  elif not_after is not None:
1047
    remaining_days = int((not_after - now) / (24 * 3600))
1048

    
1049
    msg = ("Certificate %s expires in %d days" % (filename, remaining_days))
1050

    
1051
    if remaining_days <= error_days:
1052
      return (LUVerifyCluster.ETYPE_ERROR, msg)
1053

    
1054
    if remaining_days <= warn_days:
1055
      return (LUVerifyCluster.ETYPE_WARNING, msg)
1056

    
1057
  return (None, None)
1058

    
1059

    
1060
def _VerifyCertificate(filename):
1061
  """Verifies a certificate for LUVerifyCluster.
1062

1063
  @type filename: string
1064
  @param filename: Path to PEM file
1065

1066
  """
1067
  try:
1068
    cert = OpenSSL.crypto.load_certificate(OpenSSL.crypto.FILETYPE_PEM,
1069
                                           utils.ReadFile(filename))
1070
  except Exception, err: # pylint: disable-msg=W0703
1071
    return (LUVerifyCluster.ETYPE_ERROR,
1072
            "Failed to load X509 certificate %s: %s" % (filename, err))
1073

    
1074
  # Depending on the pyOpenSSL version, this can just return (None, None)
1075
  (not_before, not_after) = utils.GetX509CertValidity(cert)
1076

    
1077
  return _VerifyCertificateInner(filename, cert.has_expired(),
1078
                                 not_before, not_after, time.time())
1079

    
1080

    
1081
class LUVerifyCluster(LogicalUnit):
1082
  """Verifies the cluster status.
1083

1084
  """
1085
  HPATH = "cluster-verify"
1086
  HTYPE = constants.HTYPE_CLUSTER
1087
  _OP_REQP = ["skip_checks", "verbose", "error_codes", "debug_simulate_errors"]
1088
  REQ_BGL = False
1089

    
1090
  TCLUSTER = "cluster"
1091
  TNODE = "node"
1092
  TINSTANCE = "instance"
1093

    
1094
  ECLUSTERCFG = (TCLUSTER, "ECLUSTERCFG")
1095
  ECLUSTERCERT = (TCLUSTER, "ECLUSTERCERT")
1096
  EINSTANCEBADNODE = (TINSTANCE, "EINSTANCEBADNODE")
1097
  EINSTANCEDOWN = (TINSTANCE, "EINSTANCEDOWN")
1098
  EINSTANCELAYOUT = (TINSTANCE, "EINSTANCELAYOUT")
1099
  EINSTANCEMISSINGDISK = (TINSTANCE, "EINSTANCEMISSINGDISK")
1100
  EINSTANCEMISSINGDISK = (TINSTANCE, "EINSTANCEMISSINGDISK")
1101
  EINSTANCEWRONGNODE = (TINSTANCE, "EINSTANCEWRONGNODE")
1102
  ENODEDRBD = (TNODE, "ENODEDRBD")
1103
  ENODEFILECHECK = (TNODE, "ENODEFILECHECK")
1104
  ENODEHOOKS = (TNODE, "ENODEHOOKS")
1105
  ENODEHV = (TNODE, "ENODEHV")
1106
  ENODELVM = (TNODE, "ENODELVM")
1107
  ENODEN1 = (TNODE, "ENODEN1")
1108
  ENODENET = (TNODE, "ENODENET")
1109
  ENODEORPHANINSTANCE = (TNODE, "ENODEORPHANINSTANCE")
1110
  ENODEORPHANLV = (TNODE, "ENODEORPHANLV")
1111
  ENODERPC = (TNODE, "ENODERPC")
1112
  ENODESSH = (TNODE, "ENODESSH")
1113
  ENODEVERSION = (TNODE, "ENODEVERSION")
1114
  ENODESETUP = (TNODE, "ENODESETUP")
1115
  ENODETIME = (TNODE, "ENODETIME")
1116

    
1117
  ETYPE_FIELD = "code"
1118
  ETYPE_ERROR = "ERROR"
1119
  ETYPE_WARNING = "WARNING"
1120

    
1121
  class NodeImage(object):
1122
    """A class representing the logical and physical status of a node.
1123

1124
    @ivar volumes: a structure as returned from
1125
        L{ganeti.backend.GetVolumeList} (runtime)
1126
    @ivar instances: a list of running instances (runtime)
1127
    @ivar pinst: list of configured primary instances (config)
1128
    @ivar sinst: list of configured secondary instances (config)
1129
    @ivar sbp: diction of {secondary-node: list of instances} of all peers
1130
        of this node (config)
1131
    @ivar mfree: free memory, as reported by hypervisor (runtime)
1132
    @ivar dfree: free disk, as reported by the node (runtime)
1133
    @ivar offline: the offline status (config)
1134
    @type rpc_fail: boolean
1135
    @ivar rpc_fail: whether the RPC verify call was successfull (overall,
1136
        not whether the individual keys were correct) (runtime)
1137
    @type lvm_fail: boolean
1138
    @ivar lvm_fail: whether the RPC call didn't return valid LVM data
1139
    @type hyp_fail: boolean
1140
    @ivar hyp_fail: whether the RPC call didn't return the instance list
1141
    @type ghost: boolean
1142
    @ivar ghost: whether this is a known node or not (config)
1143

1144
    """
1145
    def __init__(self, offline=False):
1146
      self.volumes = {}
1147
      self.instances = []
1148
      self.pinst = []
1149
      self.sinst = []
1150
      self.sbp = {}
1151
      self.mfree = 0
1152
      self.dfree = 0
1153
      self.offline = offline
1154
      self.rpc_fail = False
1155
      self.lvm_fail = False
1156
      self.hyp_fail = False
1157
      self.ghost = False
1158

    
1159
  def ExpandNames(self):
1160
    self.needed_locks = {
1161
      locking.LEVEL_NODE: locking.ALL_SET,
1162
      locking.LEVEL_INSTANCE: locking.ALL_SET,
1163
    }
1164
    self.share_locks = dict.fromkeys(locking.LEVELS, 1)
1165

    
1166
  def _Error(self, ecode, item, msg, *args, **kwargs):
1167
    """Format an error message.
1168

1169
    Based on the opcode's error_codes parameter, either format a
1170
    parseable error code, or a simpler error string.
1171

1172
    This must be called only from Exec and functions called from Exec.
1173

1174
    """
1175
    ltype = kwargs.get(self.ETYPE_FIELD, self.ETYPE_ERROR)
1176
    itype, etxt = ecode
1177
    # first complete the msg
1178
    if args:
1179
      msg = msg % args
1180
    # then format the whole message
1181
    if self.op.error_codes:
1182
      msg = "%s:%s:%s:%s:%s" % (ltype, etxt, itype, item, msg)
1183
    else:
1184
      if item:
1185
        item = " " + item
1186
      else:
1187
        item = ""
1188
      msg = "%s: %s%s: %s" % (ltype, itype, item, msg)
1189
    # and finally report it via the feedback_fn
1190
    self._feedback_fn("  - %s" % msg)
1191

    
1192
  def _ErrorIf(self, cond, *args, **kwargs):
1193
    """Log an error message if the passed condition is True.
1194

1195
    """
1196
    cond = bool(cond) or self.op.debug_simulate_errors
1197
    if cond:
1198
      self._Error(*args, **kwargs)
1199
    # do not mark the operation as failed for WARN cases only
1200
    if kwargs.get(self.ETYPE_FIELD, self.ETYPE_ERROR) == self.ETYPE_ERROR:
1201
      self.bad = self.bad or cond
1202

    
1203
  def _VerifyNode(self, ninfo, nresult):
1204
    """Run multiple tests against a node.
1205

1206
    Test list:
1207

1208
      - compares ganeti version
1209
      - checks vg existence and size > 20G
1210
      - checks config file checksum
1211
      - checks ssh to other nodes
1212

1213
    @type ninfo: L{objects.Node}
1214
    @param ninfo: the node to check
1215
    @param nresult: the results from the node
1216
    @rtype: boolean
1217
    @return: whether overall this call was successful (and we can expect
1218
         reasonable values in the respose)
1219

1220
    """
1221
    node = ninfo.name
1222
    _ErrorIf = self._ErrorIf # pylint: disable-msg=C0103
1223

    
1224
    # main result, nresult should be a non-empty dict
1225
    test = not nresult or not isinstance(nresult, dict)
1226
    _ErrorIf(test, self.ENODERPC, node,
1227
                  "unable to verify node: no data returned")
1228
    if test:
1229
      return False
1230

    
1231
    # compares ganeti version
1232
    local_version = constants.PROTOCOL_VERSION
1233
    remote_version = nresult.get("version", None)
1234
    test = not (remote_version and
1235
                isinstance(remote_version, (list, tuple)) and
1236
                len(remote_version) == 2)
1237
    _ErrorIf(test, self.ENODERPC, node,
1238
             "connection to node returned invalid data")
1239
    if test:
1240
      return False
1241

    
1242
    test = local_version != remote_version[0]
1243
    _ErrorIf(test, self.ENODEVERSION, node,
1244
             "incompatible protocol versions: master %s,"
1245
             " node %s", local_version, remote_version[0])
1246
    if test:
1247
      return False
1248

    
1249
    # node seems compatible, we can actually try to look into its results
1250

    
1251
    # full package version
1252
    self._ErrorIf(constants.RELEASE_VERSION != remote_version[1],
1253
                  self.ENODEVERSION, node,
1254
                  "software version mismatch: master %s, node %s",
1255
                  constants.RELEASE_VERSION, remote_version[1],
1256
                  code=self.ETYPE_WARNING)
1257

    
1258
    hyp_result = nresult.get(constants.NV_HYPERVISOR, None)
1259
    if isinstance(hyp_result, dict):
1260
      for hv_name, hv_result in hyp_result.iteritems():
1261
        test = hv_result is not None
1262
        _ErrorIf(test, self.ENODEHV, node,
1263
                 "hypervisor %s verify failure: '%s'", hv_name, hv_result)
1264

    
1265

    
1266
    test = nresult.get(constants.NV_NODESETUP,
1267
                           ["Missing NODESETUP results"])
1268
    _ErrorIf(test, self.ENODESETUP, node, "node setup error: %s",
1269
             "; ".join(test))
1270

    
1271
    return True
1272

    
1273
  def _VerifyNodeTime(self, ninfo, nresult,
1274
                      nvinfo_starttime, nvinfo_endtime):
1275
    """Check the node time.
1276

1277
    @type ninfo: L{objects.Node}
1278
    @param ninfo: the node to check
1279
    @param nresult: the remote results for the node
1280
    @param nvinfo_starttime: the start time of the RPC call
1281
    @param nvinfo_endtime: the end time of the RPC call
1282

1283
    """
1284
    node = ninfo.name
1285
    _ErrorIf = self._ErrorIf # pylint: disable-msg=C0103
1286

    
1287
    ntime = nresult.get(constants.NV_TIME, None)
1288
    try:
1289
      ntime_merged = utils.MergeTime(ntime)
1290
    except (ValueError, TypeError):
1291
      _ErrorIf(True, self.ENODETIME, node, "Node returned invalid time")
1292
      return
1293

    
1294
    if ntime_merged < (nvinfo_starttime - constants.NODE_MAX_CLOCK_SKEW):
1295
      ntime_diff = "%.01fs" % abs(nvinfo_starttime - ntime_merged)
1296
    elif ntime_merged > (nvinfo_endtime + constants.NODE_MAX_CLOCK_SKEW):
1297
      ntime_diff = "%.01fs" % abs(ntime_merged - nvinfo_endtime)
1298
    else:
1299
      ntime_diff = None
1300

    
1301
    _ErrorIf(ntime_diff is not None, self.ENODETIME, node,
1302
             "Node time diverges by at least %s from master node time",
1303
             ntime_diff)
1304

    
1305
  def _VerifyNodeLVM(self, ninfo, nresult, vg_name):
1306
    """Check the node time.
1307

1308
    @type ninfo: L{objects.Node}
1309
    @param ninfo: the node to check
1310
    @param nresult: the remote results for the node
1311
    @param vg_name: the configured VG name
1312

1313
    """
1314
    if vg_name is None:
1315
      return
1316

    
1317
    node = ninfo.name
1318
    _ErrorIf = self._ErrorIf # pylint: disable-msg=C0103
1319

    
1320
    # checks vg existence and size > 20G
1321
    vglist = nresult.get(constants.NV_VGLIST, None)
1322
    test = not vglist
1323
    _ErrorIf(test, self.ENODELVM, node, "unable to check volume groups")
1324
    if not test:
1325
      vgstatus = utils.CheckVolumeGroupSize(vglist, vg_name,
1326
                                            constants.MIN_VG_SIZE)
1327
      _ErrorIf(vgstatus, self.ENODELVM, node, vgstatus)
1328

    
1329
    # check pv names
1330
    pvlist = nresult.get(constants.NV_PVLIST, None)
1331
    test = pvlist is None
1332
    _ErrorIf(test, self.ENODELVM, node, "Can't get PV list from node")
1333
    if not test:
1334
      # check that ':' is not present in PV names, since it's a
1335
      # special character for lvcreate (denotes the range of PEs to
1336
      # use on the PV)
1337
      for _, pvname, owner_vg in pvlist:
1338
        test = ":" in pvname
1339
        _ErrorIf(test, self.ENODELVM, node, "Invalid character ':' in PV"
1340
                 " '%s' of VG '%s'", pvname, owner_vg)
1341

    
1342
  def _VerifyNodeNetwork(self, ninfo, nresult):
1343
    """Check the node time.
1344

1345
    @type ninfo: L{objects.Node}
1346
    @param ninfo: the node to check
1347
    @param nresult: the remote results for the node
1348

1349
    """
1350
    node = ninfo.name
1351
    _ErrorIf = self._ErrorIf # pylint: disable-msg=C0103
1352

    
1353
    test = constants.NV_NODELIST not in nresult
1354
    _ErrorIf(test, self.ENODESSH, node,
1355
             "node hasn't returned node ssh connectivity data")
1356
    if not test:
1357
      if nresult[constants.NV_NODELIST]:
1358
        for a_node, a_msg in nresult[constants.NV_NODELIST].items():
1359
          _ErrorIf(True, self.ENODESSH, node,
1360
                   "ssh communication with node '%s': %s", a_node, a_msg)
1361

    
1362
    test = constants.NV_NODENETTEST not in nresult
1363
    _ErrorIf(test, self.ENODENET, node,
1364
             "node hasn't returned node tcp connectivity data")
1365
    if not test:
1366
      if nresult[constants.NV_NODENETTEST]:
1367
        nlist = utils.NiceSort(nresult[constants.NV_NODENETTEST].keys())
1368
        for anode in nlist:
1369
          _ErrorIf(True, self.ENODENET, node,
1370
                   "tcp communication with node '%s': %s",
1371
                   anode, nresult[constants.NV_NODENETTEST][anode])
1372

    
1373
  def _VerifyInstance(self, instance, instanceconfig, node_image):
1374
    """Verify an instance.
1375

1376
    This function checks to see if the required block devices are
1377
    available on the instance's node.
1378

1379
    """
1380
    _ErrorIf = self._ErrorIf # pylint: disable-msg=C0103
1381
    node_current = instanceconfig.primary_node
1382

    
1383
    node_vol_should = {}
1384
    instanceconfig.MapLVsByNode(node_vol_should)
1385

    
1386
    for node in node_vol_should:
1387
      n_img = node_image[node]
1388
      if n_img.offline or n_img.rpc_fail or n_img.lvm_fail:
1389
        # ignore missing volumes on offline or broken nodes
1390
        continue
1391
      for volume in node_vol_should[node]:
1392
        test = volume not in n_img.volumes
1393
        _ErrorIf(test, self.EINSTANCEMISSINGDISK, instance,
1394
                 "volume %s missing on node %s", volume, node)
1395

    
1396
    if instanceconfig.admin_up:
1397
      pri_img = node_image[node_current]
1398
      test = instance not in pri_img.instances and not pri_img.offline
1399
      _ErrorIf(test, self.EINSTANCEDOWN, instance,
1400
               "instance not running on its primary node %s",
1401
               node_current)
1402

    
1403
    for node, n_img in node_image.items():
1404
      if (not node == node_current):
1405
        test = instance in n_img.instances
1406
        _ErrorIf(test, self.EINSTANCEWRONGNODE, instance,
1407
                 "instance should not run on node %s", node)
1408

    
1409
  def _VerifyOrphanVolumes(self, node_vol_should, node_image):
1410
    """Verify if there are any unknown volumes in the cluster.
1411

1412
    The .os, .swap and backup volumes are ignored. All other volumes are
1413
    reported as unknown.
1414

1415
    """
1416
    for node, n_img in node_image.items():
1417
      if n_img.offline or n_img.rpc_fail or n_img.lvm_fail:
1418
        # skip non-healthy nodes
1419
        continue
1420
      for volume in n_img.volumes:
1421
        test = (node not in node_vol_should or
1422
                volume not in node_vol_should[node])
1423
        self._ErrorIf(test, self.ENODEORPHANLV, node,
1424
                      "volume %s is unknown", volume)
1425

    
1426
  def _VerifyOrphanInstances(self, instancelist, node_image):
1427
    """Verify the list of running instances.
1428

1429
    This checks what instances are running but unknown to the cluster.
1430

1431
    """
1432
    for node, n_img in node_image.items():
1433
      for o_inst in n_img.instances:
1434
        test = o_inst not in instancelist
1435
        self._ErrorIf(test, self.ENODEORPHANINSTANCE, node,
1436
                      "instance %s on node %s should not exist", o_inst, node)
1437

    
1438
  def _VerifyNPlusOneMemory(self, node_image, instance_cfg):
1439
    """Verify N+1 Memory Resilience.
1440

1441
    Check that if one single node dies we can still start all the
1442
    instances it was primary for.
1443

1444
    """
1445
    for node, n_img in node_image.items():
1446
      # This code checks that every node which is now listed as
1447
      # secondary has enough memory to host all instances it is
1448
      # supposed to should a single other node in the cluster fail.
1449
      # FIXME: not ready for failover to an arbitrary node
1450
      # FIXME: does not support file-backed instances
1451
      # WARNING: we currently take into account down instances as well
1452
      # as up ones, considering that even if they're down someone
1453
      # might want to start them even in the event of a node failure.
1454
      for prinode, instances in n_img.sbp.items():
1455
        needed_mem = 0
1456
        for instance in instances:
1457
          bep = self.cfg.GetClusterInfo().FillBE(instance_cfg[instance])
1458
          if bep[constants.BE_AUTO_BALANCE]:
1459
            needed_mem += bep[constants.BE_MEMORY]
1460
        test = n_img.mfree < needed_mem
1461
        self._ErrorIf(test, self.ENODEN1, node,
1462
                      "not enough memory on to accommodate"
1463
                      " failovers should peer node %s fail", prinode)
1464

    
1465
  def _VerifyNodeFiles(self, ninfo, nresult, file_list, local_cksum,
1466
                       master_files):
1467
    """Verifies and computes the node required file checksums.
1468

1469
    @type ninfo: L{objects.Node}
1470
    @param ninfo: the node to check
1471
    @param nresult: the remote results for the node
1472
    @param file_list: required list of files
1473
    @param local_cksum: dictionary of local files and their checksums
1474
    @param master_files: list of files that only masters should have
1475

1476
    """
1477
    node = ninfo.name
1478
    _ErrorIf = self._ErrorIf # pylint: disable-msg=C0103
1479

    
1480
    remote_cksum = nresult.get(constants.NV_FILELIST, None)
1481
    test = not isinstance(remote_cksum, dict)
1482
    _ErrorIf(test, self.ENODEFILECHECK, node,
1483
             "node hasn't returned file checksum data")
1484
    if test:
1485
      return
1486

    
1487
    for file_name in file_list:
1488
      node_is_mc = ninfo.master_candidate
1489
      must_have = (file_name not in master_files) or node_is_mc
1490
      # missing
1491
      test1 = file_name not in remote_cksum
1492
      # invalid checksum
1493
      test2 = not test1 and remote_cksum[file_name] != local_cksum[file_name]
1494
      # existing and good
1495
      test3 = not test1 and remote_cksum[file_name] == local_cksum[file_name]
1496
      _ErrorIf(test1 and must_have, self.ENODEFILECHECK, node,
1497
               "file '%s' missing", file_name)
1498
      _ErrorIf(test2 and must_have, self.ENODEFILECHECK, node,
1499
               "file '%s' has wrong checksum", file_name)
1500
      # not candidate and this is not a must-have file
1501
      _ErrorIf(test2 and not must_have, self.ENODEFILECHECK, node,
1502
               "file '%s' should not exist on non master"
1503
               " candidates (and the file is outdated)", file_name)
1504
      # all good, except non-master/non-must have combination
1505
      _ErrorIf(test3 and not must_have, self.ENODEFILECHECK, node,
1506
               "file '%s' should not exist"
1507
               " on non master candidates", file_name)
1508

    
1509
  def _VerifyNodeDrbd(self, ninfo, nresult, instanceinfo, drbd_map):
1510
    """Verifies and the node DRBD status.
1511

1512
    @type ninfo: L{objects.Node}
1513
    @param ninfo: the node to check
1514
    @param nresult: the remote results for the node
1515
    @param instanceinfo: the dict of instances
1516
    @param drbd_map: the DRBD map as returned by
1517
        L{ganeti.config.ConfigWriter.ComputeDRBDMap}
1518

1519
    """
1520
    node = ninfo.name
1521
    _ErrorIf = self._ErrorIf # pylint: disable-msg=C0103
1522

    
1523
    # compute the DRBD minors
1524
    node_drbd = {}
1525
    for minor, instance in drbd_map[node].items():
1526
      test = instance not in instanceinfo
1527
      _ErrorIf(test, self.ECLUSTERCFG, None,
1528
               "ghost instance '%s' in temporary DRBD map", instance)
1529
        # ghost instance should not be running, but otherwise we
1530
        # don't give double warnings (both ghost instance and
1531
        # unallocated minor in use)
1532
      if test:
1533
        node_drbd[minor] = (instance, False)
1534
      else:
1535
        instance = instanceinfo[instance]
1536
        node_drbd[minor] = (instance.name, instance.admin_up)
1537

    
1538
    # and now check them
1539
    used_minors = nresult.get(constants.NV_DRBDLIST, [])
1540
    test = not isinstance(used_minors, (tuple, list))
1541
    _ErrorIf(test, self.ENODEDRBD, node,
1542
             "cannot parse drbd status file: %s", str(used_minors))
1543
    if test:
1544
      # we cannot check drbd status
1545
      return
1546

    
1547
    for minor, (iname, must_exist) in node_drbd.items():
1548
      test = minor not in used_minors and must_exist
1549
      _ErrorIf(test, self.ENODEDRBD, node,
1550
               "drbd minor %d of instance %s is not active", minor, iname)
1551
    for minor in used_minors:
1552
      test = minor not in node_drbd
1553
      _ErrorIf(test, self.ENODEDRBD, node,
1554
               "unallocated drbd minor %d is in use", minor)
1555

    
1556
  def _UpdateNodeVolumes(self, ninfo, nresult, nimg, vg_name):
1557
    """Verifies and updates the node volume data.
1558

1559
    This function will update a L{NodeImage}'s internal structures
1560
    with data from the remote call.
1561

1562
    @type ninfo: L{objects.Node}
1563
    @param ninfo: the node to check
1564
    @param nresult: the remote results for the node
1565
    @param nimg: the node image object
1566
    @param vg_name: the configured VG name
1567

1568
    """
1569
    node = ninfo.name
1570
    _ErrorIf = self._ErrorIf # pylint: disable-msg=C0103
1571

    
1572
    nimg.lvm_fail = True
1573
    lvdata = nresult.get(constants.NV_LVLIST, "Missing LV data")
1574
    if vg_name is None:
1575
      pass
1576
    elif isinstance(lvdata, basestring):
1577
      _ErrorIf(True, self.ENODELVM, node, "LVM problem on node: %s",
1578
               utils.SafeEncode(lvdata))
1579
    elif not isinstance(lvdata, dict):
1580
      _ErrorIf(True, self.ENODELVM, node, "rpc call to node failed (lvlist)")
1581
    else:
1582
      nimg.volumes = lvdata
1583
      nimg.lvm_fail = False
1584

    
1585
  def _UpdateNodeInstances(self, ninfo, nresult, nimg):
1586
    """Verifies and updates the node instance list.
1587

1588
    If the listing was successful, then updates this node's instance
1589
    list. Otherwise, it marks the RPC call as failed for the instance
1590
    list key.
1591

1592
    @type ninfo: L{objects.Node}
1593
    @param ninfo: the node to check
1594
    @param nresult: the remote results for the node
1595
    @param nimg: the node image object
1596

1597
    """
1598
    idata = nresult.get(constants.NV_INSTANCELIST, None)
1599
    test = not isinstance(idata, list)
1600
    self._ErrorIf(test, self.ENODEHV, ninfo.name, "rpc call to node failed"
1601
                  " (instancelist): %s", utils.SafeEncode(str(idata)))
1602
    if test:
1603
      nimg.hyp_fail = True
1604
    else:
1605
      nimg.instances = idata
1606

    
1607
  def _UpdateNodeInfo(self, ninfo, nresult, nimg, vg_name):
1608
    """Verifies and computes a node information map
1609

1610
    @type ninfo: L{objects.Node}
1611
    @param ninfo: the node to check
1612
    @param nresult: the remote results for the node
1613
    @param nimg: the node image object
1614
    @param vg_name: the configured VG name
1615

1616
    """
1617
    node = ninfo.name
1618
    _ErrorIf = self._ErrorIf # pylint: disable-msg=C0103
1619

    
1620
    # try to read free memory (from the hypervisor)
1621
    hv_info = nresult.get(constants.NV_HVINFO, None)
1622
    test = not isinstance(hv_info, dict) or "memory_free" not in hv_info
1623
    _ErrorIf(test, self.ENODEHV, node, "rpc call to node failed (hvinfo)")
1624
    if not test:
1625
      try:
1626
        nimg.mfree = int(hv_info["memory_free"])
1627
      except (ValueError, TypeError):
1628
        _ErrorIf(True, self.ENODERPC, node,
1629
                 "node returned invalid nodeinfo, check hypervisor")
1630

    
1631
    # FIXME: devise a free space model for file based instances as well
1632
    if vg_name is not None:
1633
      test = (constants.NV_VGLIST not in nresult or
1634
              vg_name not in nresult[constants.NV_VGLIST])
1635
      _ErrorIf(test, self.ENODELVM, node,
1636
               "node didn't return data for the volume group '%s'"
1637
               " - it is either missing or broken", vg_name)
1638
      if not test:
1639
        try:
1640
          nimg.dfree = int(nresult[constants.NV_VGLIST][vg_name])
1641
        except (ValueError, TypeError):
1642
          _ErrorIf(True, self.ENODERPC, node,
1643
                   "node returned invalid LVM info, check LVM status")
1644

    
1645
  def CheckPrereq(self):
1646
    """Check prerequisites.
1647

1648
    Transform the list of checks we're going to skip into a set and check that
1649
    all its members are valid.
1650

1651
    """
1652
    self.skip_set = frozenset(self.op.skip_checks)
1653
    if not constants.VERIFY_OPTIONAL_CHECKS.issuperset(self.skip_set):
1654
      raise errors.OpPrereqError("Invalid checks to be skipped specified",
1655
                                 errors.ECODE_INVAL)
1656

    
1657
  def BuildHooksEnv(self):
1658
    """Build hooks env.
1659

1660
    Cluster-Verify hooks just ran in the post phase and their failure makes
1661
    the output be logged in the verify output and the verification to fail.
1662

1663
    """
1664
    all_nodes = self.cfg.GetNodeList()
1665
    env = {
1666
      "CLUSTER_TAGS": " ".join(self.cfg.GetClusterInfo().GetTags())
1667
      }
1668
    for node in self.cfg.GetAllNodesInfo().values():
1669
      env["NODE_TAGS_%s" % node.name] = " ".join(node.GetTags())
1670

    
1671
    return env, [], all_nodes
1672

    
1673
  def Exec(self, feedback_fn):
1674
    """Verify integrity of cluster, performing various test on nodes.
1675

1676
    """
1677
    self.bad = False
1678
    _ErrorIf = self._ErrorIf # pylint: disable-msg=C0103
1679
    verbose = self.op.verbose
1680
    self._feedback_fn = feedback_fn
1681
    feedback_fn("* Verifying global settings")
1682
    for msg in self.cfg.VerifyConfig():
1683
      _ErrorIf(True, self.ECLUSTERCFG, None, msg)
1684

    
1685
    # Check the cluster certificates
1686
    for cert_filename in constants.ALL_CERT_FILES:
1687
      (errcode, msg) = _VerifyCertificate(cert_filename)
1688
      _ErrorIf(errcode, self.ECLUSTERCERT, None, msg, code=errcode)
1689

    
1690
    vg_name = self.cfg.GetVGName()
1691
    hypervisors = self.cfg.GetClusterInfo().enabled_hypervisors
1692
    nodelist = utils.NiceSort(self.cfg.GetNodeList())
1693
    nodeinfo = [self.cfg.GetNodeInfo(nname) for nname in nodelist]
1694
    instancelist = utils.NiceSort(self.cfg.GetInstanceList())
1695
    instanceinfo = dict((iname, self.cfg.GetInstanceInfo(iname))
1696
                        for iname in instancelist)
1697
    i_non_redundant = [] # Non redundant instances
1698
    i_non_a_balanced = [] # Non auto-balanced instances
1699
    n_offline = 0 # Count of offline nodes
1700
    n_drained = 0 # Count of nodes being drained
1701
    node_vol_should = {}
1702

    
1703
    # FIXME: verify OS list
1704
    # do local checksums
1705
    master_files = [constants.CLUSTER_CONF_FILE]
1706

    
1707
    file_names = ssconf.SimpleStore().GetFileList()
1708
    file_names.extend(constants.ALL_CERT_FILES)
1709
    file_names.extend(master_files)
1710

    
1711
    local_checksums = utils.FingerprintFiles(file_names)
1712

    
1713
    feedback_fn("* Gathering data (%d nodes)" % len(nodelist))
1714
    node_verify_param = {
1715
      constants.NV_FILELIST: file_names,
1716
      constants.NV_NODELIST: [node.name for node in nodeinfo
1717
                              if not node.offline],
1718
      constants.NV_HYPERVISOR: hypervisors,
1719
      constants.NV_NODENETTEST: [(node.name, node.primary_ip,
1720
                                  node.secondary_ip) for node in nodeinfo
1721
                                 if not node.offline],
1722
      constants.NV_INSTANCELIST: hypervisors,
1723
      constants.NV_VERSION: None,
1724
      constants.NV_HVINFO: self.cfg.GetHypervisorType(),
1725
      constants.NV_NODESETUP: None,
1726
      constants.NV_TIME: None,
1727
      }
1728

    
1729
    if vg_name is not None:
1730
      node_verify_param[constants.NV_VGLIST] = None
1731
      node_verify_param[constants.NV_LVLIST] = vg_name
1732
      node_verify_param[constants.NV_PVLIST] = [vg_name]
1733
      node_verify_param[constants.NV_DRBDLIST] = None
1734

    
1735
    # Build our expected cluster state
1736
    node_image = dict((node.name, self.NodeImage(offline=node.offline))
1737
                      for node in nodeinfo)
1738

    
1739
    for instance in instancelist:
1740
      inst_config = instanceinfo[instance]
1741

    
1742
      for nname in inst_config.all_nodes:
1743
        if nname not in node_image:
1744
          # ghost node
1745
          gnode = self.NodeImage()
1746
          gnode.ghost = True
1747
          node_image[nname] = gnode
1748

    
1749
      inst_config.MapLVsByNode(node_vol_should)
1750

    
1751
      pnode = inst_config.primary_node
1752
      node_image[pnode].pinst.append(instance)
1753

    
1754
      for snode in inst_config.secondary_nodes:
1755
        nimg = node_image[snode]
1756
        nimg.sinst.append(instance)
1757
        if pnode not in nimg.sbp:
1758
          nimg.sbp[pnode] = []
1759
        nimg.sbp[pnode].append(instance)
1760

    
1761
    # At this point, we have the in-memory data structures complete,
1762
    # except for the runtime information, which we'll gather next
1763

    
1764
    # Due to the way our RPC system works, exact response times cannot be
1765
    # guaranteed (e.g. a broken node could run into a timeout). By keeping the
1766
    # time before and after executing the request, we can at least have a time
1767
    # window.
1768
    nvinfo_starttime = time.time()
1769
    all_nvinfo = self.rpc.call_node_verify(nodelist, node_verify_param,
1770
                                           self.cfg.GetClusterName())
1771
    nvinfo_endtime = time.time()
1772

    
1773
    cluster = self.cfg.GetClusterInfo()
1774
    master_node = self.cfg.GetMasterNode()
1775
    all_drbd_map = self.cfg.ComputeDRBDMap()
1776

    
1777
    feedback_fn("* Verifying node status")
1778
    for node_i in nodeinfo:
1779
      node = node_i.name
1780
      nimg = node_image[node]
1781

    
1782
      if node_i.offline:
1783
        if verbose:
1784
          feedback_fn("* Skipping offline node %s" % (node,))
1785
        n_offline += 1
1786
        continue
1787

    
1788
      if node == master_node:
1789
        ntype = "master"
1790
      elif node_i.master_candidate:
1791
        ntype = "master candidate"
1792
      elif node_i.drained:
1793
        ntype = "drained"
1794
        n_drained += 1
1795
      else:
1796
        ntype = "regular"
1797
      if verbose:
1798
        feedback_fn("* Verifying node %s (%s)" % (node, ntype))
1799

    
1800
      msg = all_nvinfo[node].fail_msg
1801
      _ErrorIf(msg, self.ENODERPC, node, "while contacting node: %s", msg)
1802
      if msg:
1803
        nimg.rpc_fail = True
1804
        continue
1805

    
1806
      nresult = all_nvinfo[node].payload
1807

    
1808
      nimg.call_ok = self._VerifyNode(node_i, nresult)
1809
      self._VerifyNodeNetwork(node_i, nresult)
1810
      self._VerifyNodeLVM(node_i, nresult, vg_name)
1811
      self._VerifyNodeFiles(node_i, nresult, file_names, local_checksums,
1812
                            master_files)
1813
      self._VerifyNodeDrbd(node_i, nresult, instanceinfo, all_drbd_map)
1814
      self._VerifyNodeTime(node_i, nresult, nvinfo_starttime, nvinfo_endtime)
1815

    
1816
      self._UpdateNodeVolumes(node_i, nresult, nimg, vg_name)
1817
      self._UpdateNodeInstances(node_i, nresult, nimg)
1818
      self._UpdateNodeInfo(node_i, nresult, nimg, vg_name)
1819

    
1820
    feedback_fn("* Verifying instance status")
1821
    for instance in instancelist:
1822
      if verbose:
1823
        feedback_fn("* Verifying instance %s" % instance)
1824
      inst_config = instanceinfo[instance]
1825
      self._VerifyInstance(instance, inst_config, node_image)
1826
      inst_nodes_offline = []
1827

    
1828
      pnode = inst_config.primary_node
1829
      pnode_img = node_image[pnode]
1830
      _ErrorIf(pnode_img.rpc_fail and not pnode_img.offline,
1831
               self.ENODERPC, pnode, "instance %s, connection to"
1832
               " primary node failed", instance)
1833

    
1834
      if pnode_img.offline:
1835
        inst_nodes_offline.append(pnode)
1836

    
1837
      # If the instance is non-redundant we cannot survive losing its primary
1838
      # node, so we are not N+1 compliant. On the other hand we have no disk
1839
      # templates with more than one secondary so that situation is not well
1840
      # supported either.
1841
      # FIXME: does not support file-backed instances
1842
      if not inst_config.secondary_nodes:
1843
        i_non_redundant.append(instance)
1844
      _ErrorIf(len(inst_config.secondary_nodes) > 1, self.EINSTANCELAYOUT,
1845
               instance, "instance has multiple secondary nodes: %s",
1846
               utils.CommaJoin(inst_config.secondary_nodes),
1847
               code=self.ETYPE_WARNING)
1848

    
1849
      if not cluster.FillBE(inst_config)[constants.BE_AUTO_BALANCE]:
1850
        i_non_a_balanced.append(instance)
1851

    
1852
      for snode in inst_config.secondary_nodes:
1853
        s_img = node_image[snode]
1854
        _ErrorIf(s_img.rpc_fail and not s_img.offline, self.ENODERPC, snode,
1855
                 "instance %s, connection to secondary node failed", instance)
1856

    
1857
        if s_img.offline:
1858
          inst_nodes_offline.append(snode)
1859

    
1860
      # warn that the instance lives on offline nodes
1861
      _ErrorIf(inst_nodes_offline, self.EINSTANCEBADNODE, instance,
1862
               "instance lives on offline node(s) %s",
1863
               utils.CommaJoin(inst_nodes_offline))
1864
      # ... or ghost nodes
1865
      for node in inst_config.all_nodes:
1866
        _ErrorIf(node_image[node].ghost, self.EINSTANCEBADNODE, instance,
1867
                 "instance lives on ghost node %s", node)
1868

    
1869
    feedback_fn("* Verifying orphan volumes")
1870
    self._VerifyOrphanVolumes(node_vol_should, node_image)
1871

    
1872
    feedback_fn("* Verifying oprhan instances")
1873
    self._VerifyOrphanInstances(instancelist, node_image)
1874

    
1875
    if constants.VERIFY_NPLUSONE_MEM not in self.skip_set:
1876
      feedback_fn("* Verifying N+1 Memory redundancy")
1877
      self._VerifyNPlusOneMemory(node_image, instanceinfo)
1878

    
1879
    feedback_fn("* Other Notes")
1880
    if i_non_redundant:
1881
      feedback_fn("  - NOTICE: %d non-redundant instance(s) found."
1882
                  % len(i_non_redundant))
1883

    
1884
    if i_non_a_balanced:
1885
      feedback_fn("  - NOTICE: %d non-auto-balanced instance(s) found."
1886
                  % len(i_non_a_balanced))
1887

    
1888
    if n_offline:
1889
      feedback_fn("  - NOTICE: %d offline node(s) found." % n_offline)
1890

    
1891
    if n_drained:
1892
      feedback_fn("  - NOTICE: %d drained node(s) found." % n_drained)
1893

    
1894
    return not self.bad
1895

    
1896
  def HooksCallBack(self, phase, hooks_results, feedback_fn, lu_result):
1897
    """Analyze the post-hooks' result
1898

1899
    This method analyses the hook result, handles it, and sends some
1900
    nicely-formatted feedback back to the user.
1901

1902
    @param phase: one of L{constants.HOOKS_PHASE_POST} or
1903
        L{constants.HOOKS_PHASE_PRE}; it denotes the hooks phase
1904
    @param hooks_results: the results of the multi-node hooks rpc call
1905
    @param feedback_fn: function used send feedback back to the caller
1906
    @param lu_result: previous Exec result
1907
    @return: the new Exec result, based on the previous result
1908
        and hook results
1909

1910
    """
1911
    # We only really run POST phase hooks, and are only interested in
1912
    # their results
1913
    if phase == constants.HOOKS_PHASE_POST:
1914
      # Used to change hooks' output to proper indentation
1915
      indent_re = re.compile('^', re.M)
1916
      feedback_fn("* Hooks Results")
1917
      assert hooks_results, "invalid result from hooks"
1918

    
1919
      for node_name in hooks_results:
1920
        res = hooks_results[node_name]
1921
        msg = res.fail_msg
1922
        test = msg and not res.offline
1923
        self._ErrorIf(test, self.ENODEHOOKS, node_name,
1924
                      "Communication failure in hooks execution: %s", msg)
1925
        if res.offline or msg:
1926
          # No need to investigate payload if node is offline or gave an error.
1927
          # override manually lu_result here as _ErrorIf only
1928
          # overrides self.bad
1929
          lu_result = 1
1930
          continue
1931
        for script, hkr, output in res.payload:
1932
          test = hkr == constants.HKR_FAIL
1933
          self._ErrorIf(test, self.ENODEHOOKS, node_name,
1934
                        "Script %s failed, output:", script)
1935
          if test:
1936
            output = indent_re.sub('      ', output)
1937
            feedback_fn("%s" % output)
1938
            lu_result = 0
1939

    
1940
      return lu_result
1941

    
1942

    
1943
class LUVerifyDisks(NoHooksLU):
1944
  """Verifies the cluster disks status.
1945

1946
  """
1947
  _OP_REQP = []
1948
  REQ_BGL = False
1949

    
1950
  def ExpandNames(self):
1951
    self.needed_locks = {
1952
      locking.LEVEL_NODE: locking.ALL_SET,
1953
      locking.LEVEL_INSTANCE: locking.ALL_SET,
1954
    }
1955
    self.share_locks = dict.fromkeys(locking.LEVELS, 1)
1956

    
1957
  def CheckPrereq(self):
1958
    """Check prerequisites.
1959

1960
    This has no prerequisites.
1961

1962
    """
1963
    pass
1964

    
1965
  def Exec(self, feedback_fn):
1966
    """Verify integrity of cluster disks.
1967

1968
    @rtype: tuple of three items
1969
    @return: a tuple of (dict of node-to-node_error, list of instances
1970
        which need activate-disks, dict of instance: (node, volume) for
1971
        missing volumes
1972

1973
    """
1974
    result = res_nodes, res_instances, res_missing = {}, [], {}
1975

    
1976
    vg_name = self.cfg.GetVGName()
1977
    nodes = utils.NiceSort(self.cfg.GetNodeList())
1978
    instances = [self.cfg.GetInstanceInfo(name)
1979
                 for name in self.cfg.GetInstanceList()]
1980

    
1981
    nv_dict = {}
1982
    for inst in instances:
1983
      inst_lvs = {}
1984
      if (not inst.admin_up or
1985
          inst.disk_template not in constants.DTS_NET_MIRROR):
1986
        continue
1987
      inst.MapLVsByNode(inst_lvs)
1988
      # transform { iname: {node: [vol,],},} to {(node, vol): iname}
1989
      for node, vol_list in inst_lvs.iteritems():
1990
        for vol in vol_list:
1991
          nv_dict[(node, vol)] = inst
1992

    
1993
    if not nv_dict:
1994
      return result
1995

    
1996
    node_lvs = self.rpc.call_lv_list(nodes, vg_name)
1997

    
1998
    for node in nodes:
1999
      # node_volume
2000
      node_res = node_lvs[node]
2001
      if node_res.offline:
2002
        continue
2003
      msg = node_res.fail_msg
2004
      if msg:
2005
        logging.warning("Error enumerating LVs on node %s: %s", node, msg)
2006
        res_nodes[node] = msg
2007
        continue
2008

    
2009
      lvs = node_res.payload
2010
      for lv_name, (_, _, lv_online) in lvs.items():
2011
        inst = nv_dict.pop((node, lv_name), None)
2012
        if (not lv_online and inst is not None
2013
            and inst.name not in res_instances):
2014
          res_instances.append(inst.name)
2015

    
2016
    # any leftover items in nv_dict are missing LVs, let's arrange the
2017
    # data better
2018
    for key, inst in nv_dict.iteritems():
2019
      if inst.name not in res_missing:
2020
        res_missing[inst.name] = []
2021
      res_missing[inst.name].append(key)
2022

    
2023
    return result
2024

    
2025

    
2026
class LURepairDiskSizes(NoHooksLU):
2027
  """Verifies the cluster disks sizes.
2028

2029
  """
2030
  _OP_REQP = ["instances"]
2031
  REQ_BGL = False
2032

    
2033
  def ExpandNames(self):
2034
    if not isinstance(self.op.instances, list):
2035
      raise errors.OpPrereqError("Invalid argument type 'instances'",
2036
                                 errors.ECODE_INVAL)
2037

    
2038
    if self.op.instances:
2039
      self.wanted_names = []
2040
      for name in self.op.instances:
2041
        full_name = _ExpandInstanceName(self.cfg, name)
2042
        self.wanted_names.append(full_name)
2043
      self.needed_locks = {
2044
        locking.LEVEL_NODE: [],
2045
        locking.LEVEL_INSTANCE: self.wanted_names,
2046
        }
2047
      self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
2048
    else:
2049
      self.wanted_names = None
2050
      self.needed_locks = {
2051
        locking.LEVEL_NODE: locking.ALL_SET,
2052
        locking.LEVEL_INSTANCE: locking.ALL_SET,
2053
        }
2054
    self.share_locks = dict(((i, 1) for i in locking.LEVELS))
2055

    
2056
  def DeclareLocks(self, level):
2057
    if level == locking.LEVEL_NODE and self.wanted_names is not None:
2058
      self._LockInstancesNodes(primary_only=True)
2059

    
2060
  def CheckPrereq(self):
2061
    """Check prerequisites.
2062

2063
    This only checks the optional instance list against the existing names.
2064

2065
    """
2066
    if self.wanted_names is None:
2067
      self.wanted_names = self.acquired_locks[locking.LEVEL_INSTANCE]
2068

    
2069
    self.wanted_instances = [self.cfg.GetInstanceInfo(name) for name
2070
                             in self.wanted_names]
2071

    
2072
  def _EnsureChildSizes(self, disk):
2073
    """Ensure children of the disk have the needed disk size.
2074

2075
    This is valid mainly for DRBD8 and fixes an issue where the
2076
    children have smaller disk size.
2077

2078
    @param disk: an L{ganeti.objects.Disk} object
2079

2080
    """
2081
    if disk.dev_type == constants.LD_DRBD8:
2082
      assert disk.children, "Empty children for DRBD8?"
2083
      fchild = disk.children[0]
2084
      mismatch = fchild.size < disk.size
2085
      if mismatch:
2086
        self.LogInfo("Child disk has size %d, parent %d, fixing",
2087
                     fchild.size, disk.size)
2088
        fchild.size = disk.size
2089

    
2090
      # and we recurse on this child only, not on the metadev
2091
      return self._EnsureChildSizes(fchild) or mismatch
2092
    else:
2093
      return False
2094

    
2095
  def Exec(self, feedback_fn):
2096
    """Verify the size of cluster disks.
2097

2098
    """
2099
    # TODO: check child disks too
2100
    # TODO: check differences in size between primary/secondary nodes
2101
    per_node_disks = {}
2102
    for instance in self.wanted_instances:
2103
      pnode = instance.primary_node
2104
      if pnode not in per_node_disks:
2105
        per_node_disks[pnode] = []
2106
      for idx, disk in enumerate(instance.disks):
2107
        per_node_disks[pnode].append((instance, idx, disk))
2108

    
2109
    changed = []
2110
    for node, dskl in per_node_disks.items():
2111
      newl = [v[2].Copy() for v in dskl]
2112
      for dsk in newl:
2113
        self.cfg.SetDiskID(dsk, node)
2114
      result = self.rpc.call_blockdev_getsizes(node, newl)
2115
      if result.fail_msg:
2116
        self.LogWarning("Failure in blockdev_getsizes call to node"
2117
                        " %s, ignoring", node)
2118
        continue
2119
      if len(result.data) != len(dskl):
2120
        self.LogWarning("Invalid result from node %s, ignoring node results",
2121
                        node)
2122
        continue
2123
      for ((instance, idx, disk), size) in zip(dskl, result.data):
2124
        if size is None:
2125
          self.LogWarning("Disk %d of instance %s did not return size"
2126
                          " information, ignoring", idx, instance.name)
2127
          continue
2128
        if not isinstance(size, (int, long)):
2129
          self.LogWarning("Disk %d of instance %s did not return valid"
2130
                          " size information, ignoring", idx, instance.name)
2131
          continue
2132
        size = size >> 20
2133
        if size != disk.size:
2134
          self.LogInfo("Disk %d of instance %s has mismatched size,"
2135
                       " correcting: recorded %d, actual %d", idx,
2136
                       instance.name, disk.size, size)
2137
          disk.size = size
2138
          self.cfg.Update(instance, feedback_fn)
2139
          changed.append((instance.name, idx, size))
2140
        if self._EnsureChildSizes(disk):
2141
          self.cfg.Update(instance, feedback_fn)
2142
          changed.append((instance.name, idx, disk.size))
2143
    return changed
2144

    
2145

    
2146
class LURenameCluster(LogicalUnit):
2147
  """Rename the cluster.
2148

2149
  """
2150
  HPATH = "cluster-rename"
2151
  HTYPE = constants.HTYPE_CLUSTER
2152
  _OP_REQP = ["name"]
2153

    
2154
  def BuildHooksEnv(self):
2155
    """Build hooks env.
2156

2157
    """
2158
    env = {
2159
      "OP_TARGET": self.cfg.GetClusterName(),
2160
      "NEW_NAME": self.op.name,
2161
      }
2162
    mn = self.cfg.GetMasterNode()
2163
    all_nodes = self.cfg.GetNodeList()
2164
    return env, [mn], all_nodes
2165

    
2166
  def CheckPrereq(self):
2167
    """Verify that the passed name is a valid one.
2168

2169
    """
2170
    hostname = utils.GetHostInfo(self.op.name)
2171

    
2172
    new_name = hostname.name
2173
    self.ip = new_ip = hostname.ip
2174
    old_name = self.cfg.GetClusterName()
2175
    old_ip = self.cfg.GetMasterIP()
2176
    if new_name == old_name and new_ip == old_ip:
2177
      raise errors.OpPrereqError("Neither the name nor the IP address of the"
2178
                                 " cluster has changed",
2179
                                 errors.ECODE_INVAL)
2180
    if new_ip != old_ip:
2181
      if utils.TcpPing(new_ip, constants.DEFAULT_NODED_PORT):
2182
        raise errors.OpPrereqError("The given cluster IP address (%s) is"
2183
                                   " reachable on the network. Aborting." %
2184
                                   new_ip, errors.ECODE_NOTUNIQUE)
2185

    
2186
    self.op.name = new_name
2187

    
2188
  def Exec(self, feedback_fn):
2189
    """Rename the cluster.
2190

2191
    """
2192
    clustername = self.op.name
2193
    ip = self.ip
2194

    
2195
    # shutdown the master IP
2196
    master = self.cfg.GetMasterNode()
2197
    result = self.rpc.call_node_stop_master(master, False)
2198
    result.Raise("Could not disable the master role")
2199

    
2200
    try:
2201
      cluster = self.cfg.GetClusterInfo()
2202
      cluster.cluster_name = clustername
2203
      cluster.master_ip = ip
2204
      self.cfg.Update(cluster, feedback_fn)
2205

    
2206
      # update the known hosts file
2207
      ssh.WriteKnownHostsFile(self.cfg, constants.SSH_KNOWN_HOSTS_FILE)
2208
      node_list = self.cfg.GetNodeList()
2209
      try:
2210
        node_list.remove(master)
2211
      except ValueError:
2212
        pass
2213
      result = self.rpc.call_upload_file(node_list,
2214
                                         constants.SSH_KNOWN_HOSTS_FILE)
2215
      for to_node, to_result in result.iteritems():
2216
        msg = to_result.fail_msg
2217
        if msg:
2218
          msg = ("Copy of file %s to node %s failed: %s" %
2219
                 (constants.SSH_KNOWN_HOSTS_FILE, to_node, msg))
2220
          self.proc.LogWarning(msg)
2221

    
2222
    finally:
2223
      result = self.rpc.call_node_start_master(master, False, False)
2224
      msg = result.fail_msg
2225
      if msg:
2226
        self.LogWarning("Could not re-enable the master role on"
2227
                        " the master, please restart manually: %s", msg)
2228

    
2229

    
2230
def _RecursiveCheckIfLVMBased(disk):
2231
  """Check if the given disk or its children are lvm-based.
2232

2233
  @type disk: L{objects.Disk}
2234
  @param disk: the disk to check
2235
  @rtype: boolean
2236
  @return: boolean indicating whether a LD_LV dev_type was found or not
2237

2238
  """
2239
  if disk.children:
2240
    for chdisk in disk.children:
2241
      if _RecursiveCheckIfLVMBased(chdisk):
2242
        return True
2243
  return disk.dev_type == constants.LD_LV
2244

    
2245

    
2246
class LUSetClusterParams(LogicalUnit):
2247
  """Change the parameters of the cluster.
2248

2249
  """
2250
  HPATH = "cluster-modify"
2251
  HTYPE = constants.HTYPE_CLUSTER
2252
  _OP_REQP = []
2253
  REQ_BGL = False
2254

    
2255
  def CheckArguments(self):
2256
    """Check parameters
2257

2258
    """
2259
    for attr in ["candidate_pool_size",
2260
                 "uid_pool", "add_uids", "remove_uids"]:
2261
      if not hasattr(self.op, attr):
2262
        setattr(self.op, attr, None)
2263

    
2264
    if self.op.candidate_pool_size is not None:
2265
      try:
2266
        self.op.candidate_pool_size = int(self.op.candidate_pool_size)
2267
      except (ValueError, TypeError), err:
2268
        raise errors.OpPrereqError("Invalid candidate_pool_size value: %s" %
2269
                                   str(err), errors.ECODE_INVAL)
2270
      if self.op.candidate_pool_size < 1:
2271
        raise errors.OpPrereqError("At least one master candidate needed",
2272
                                   errors.ECODE_INVAL)
2273

    
2274
    _CheckBooleanOpField(self.op, "maintain_node_health")
2275

    
2276
    if self.op.uid_pool:
2277
      uidpool.CheckUidPool(self.op.uid_pool)
2278

    
2279
    if self.op.add_uids:
2280
      uidpool.CheckUidPool(self.op.add_uids)
2281

    
2282
    if self.op.remove_uids:
2283
      uidpool.CheckUidPool(self.op.remove_uids)
2284

    
2285
  def ExpandNames(self):
2286
    # FIXME: in the future maybe other cluster params won't require checking on
2287
    # all nodes to be modified.
2288
    self.needed_locks = {
2289
      locking.LEVEL_NODE: locking.ALL_SET,
2290
    }
2291
    self.share_locks[locking.LEVEL_NODE] = 1
2292

    
2293
  def BuildHooksEnv(self):
2294
    """Build hooks env.
2295

2296
    """
2297
    env = {
2298
      "OP_TARGET": self.cfg.GetClusterName(),
2299
      "NEW_VG_NAME": self.op.vg_name,
2300
      }
2301
    mn = self.cfg.GetMasterNode()
2302
    return env, [mn], [mn]
2303

    
2304
  def CheckPrereq(self):
2305
    """Check prerequisites.
2306

2307
    This checks whether the given params don't conflict and
2308
    if the given volume group is valid.
2309

2310
    """
2311
    if self.op.vg_name is not None and not self.op.vg_name:
2312
      instances = self.cfg.GetAllInstancesInfo().values()
2313
      for inst in instances:
2314
        for disk in inst.disks:
2315
          if _RecursiveCheckIfLVMBased(disk):
2316
            raise errors.OpPrereqError("Cannot disable lvm storage while"
2317
                                       " lvm-based instances exist",
2318
                                       errors.ECODE_INVAL)
2319

    
2320
    node_list = self.acquired_locks[locking.LEVEL_NODE]
2321

    
2322
    # if vg_name not None, checks given volume group on all nodes
2323
    if self.op.vg_name:
2324
      vglist = self.rpc.call_vg_list(node_list)
2325
      for node in node_list:
2326
        msg = vglist[node].fail_msg
2327
        if msg:
2328
          # ignoring down node
2329
          self.LogWarning("Error while gathering data on node %s"
2330
                          " (ignoring node): %s", node, msg)
2331
          continue
2332
        vgstatus = utils.CheckVolumeGroupSize(vglist[node].payload,
2333
                                              self.op.vg_name,
2334
                                              constants.MIN_VG_SIZE)
2335
        if vgstatus:
2336
          raise errors.OpPrereqError("Error on node '%s': %s" %
2337
                                     (node, vgstatus), errors.ECODE_ENVIRON)
2338

    
2339
    self.cluster = cluster = self.cfg.GetClusterInfo()
2340
    # validate params changes
2341
    if self.op.beparams:
2342
      utils.ForceDictType(self.op.beparams, constants.BES_PARAMETER_TYPES)
2343
      self.new_beparams = objects.FillDict(
2344
        cluster.beparams[constants.PP_DEFAULT], self.op.beparams)
2345

    
2346
    if self.op.nicparams:
2347
      utils.ForceDictType(self.op.nicparams, constants.NICS_PARAMETER_TYPES)
2348
      self.new_nicparams = objects.FillDict(
2349
        cluster.nicparams[constants.PP_DEFAULT], self.op.nicparams)
2350
      objects.NIC.CheckParameterSyntax(self.new_nicparams)
2351
      nic_errors = []
2352

    
2353
      # check all instances for consistency
2354
      for instance in self.cfg.GetAllInstancesInfo().values():
2355
        for nic_idx, nic in enumerate(instance.nics):
2356
          params_copy = copy.deepcopy(nic.nicparams)
2357
          params_filled = objects.FillDict(self.new_nicparams, params_copy)
2358

    
2359
          # check parameter syntax
2360
          try:
2361
            objects.NIC.CheckParameterSyntax(params_filled)
2362
          except errors.ConfigurationError, err:
2363
            nic_errors.append("Instance %s, nic/%d: %s" %
2364
                              (instance.name, nic_idx, err))
2365

    
2366
          # if we're moving instances to routed, check that they have an ip
2367
          target_mode = params_filled[constants.NIC_MODE]
2368
          if target_mode == constants.NIC_MODE_ROUTED and not nic.ip:
2369
            nic_errors.append("Instance %s, nic/%d: routed nick with no ip" %
2370
                              (instance.name, nic_idx))
2371
      if nic_errors:
2372
        raise errors.OpPrereqError("Cannot apply the change, errors:\n%s" %
2373
                                   "\n".join(nic_errors))
2374

    
2375
    # hypervisor list/parameters
2376
    self.new_hvparams = new_hvp = objects.FillDict(cluster.hvparams, {})
2377
    if self.op.hvparams:
2378
      if not isinstance(self.op.hvparams, dict):
2379
        raise errors.OpPrereqError("Invalid 'hvparams' parameter on input",
2380
                                   errors.ECODE_INVAL)
2381
      for hv_name, hv_dict in self.op.hvparams.items():
2382
        if hv_name not in self.new_hvparams:
2383
          self.new_hvparams[hv_name] = hv_dict
2384
        else:
2385
          self.new_hvparams[hv_name].update(hv_dict)
2386

    
2387
    # os hypervisor parameters
2388
    self.new_os_hvp = objects.FillDict(cluster.os_hvp, {})
2389
    if self.op.os_hvp:
2390
      if not isinstance(self.op.os_hvp, dict):
2391
        raise errors.OpPrereqError("Invalid 'os_hvp' parameter on input",
2392
                                   errors.ECODE_INVAL)
2393
      for os_name, hvs in self.op.os_hvp.items():
2394
        if not isinstance(hvs, dict):
2395
          raise errors.OpPrereqError(("Invalid 'os_hvp' parameter on"
2396
                                      " input"), errors.ECODE_INVAL)
2397
        if os_name not in self.new_os_hvp:
2398
          self.new_os_hvp[os_name] = hvs
2399
        else:
2400
          for hv_name, hv_dict in hvs.items():
2401
            if hv_name not in self.new_os_hvp[os_name]:
2402
              self.new_os_hvp[os_name][hv_name] = hv_dict
2403
            else:
2404
              self.new_os_hvp[os_name][hv_name].update(hv_dict)
2405

    
2406
    # changes to the hypervisor list
2407
    if self.op.enabled_hypervisors is not None:
2408
      self.hv_list = self.op.enabled_hypervisors
2409
      if not self.hv_list:
2410
        raise errors.OpPrereqError("Enabled hypervisors list must contain at"
2411
                                   " least one member",
2412
                                   errors.ECODE_INVAL)
2413
      invalid_hvs = set(self.hv_list) - constants.HYPER_TYPES
2414
      if invalid_hvs:
2415
        raise errors.OpPrereqError("Enabled hypervisors contains invalid"
2416
                                   " entries: %s" %
2417
                                   utils.CommaJoin(invalid_hvs),
2418
                                   errors.ECODE_INVAL)
2419
      for hv in self.hv_list:
2420
        # if the hypervisor doesn't already exist in the cluster
2421
        # hvparams, we initialize it to empty, and then (in both
2422
        # cases) we make sure to fill the defaults, as we might not
2423
        # have a complete defaults list if the hypervisor wasn't
2424
        # enabled before
2425
        if hv not in new_hvp:
2426
          new_hvp[hv] = {}
2427
        new_hvp[hv] = objects.FillDict(constants.HVC_DEFAULTS[hv], new_hvp[hv])
2428
        utils.ForceDictType(new_hvp[hv], constants.HVS_PARAMETER_TYPES)
2429
    else:
2430
      self.hv_list = cluster.enabled_hypervisors
2431

    
2432
    if self.op.hvparams or self.op.enabled_hypervisors is not None:
2433
      # either the enabled list has changed, or the parameters have, validate
2434
      for hv_name, hv_params in self.new_hvparams.items():
2435
        if ((self.op.hvparams and hv_name in self.op.hvparams) or
2436
            (self.op.enabled_hypervisors and
2437
             hv_name in self.op.enabled_hypervisors)):
2438
          # either this is a new hypervisor, or its parameters have changed
2439
          hv_class = hypervisor.GetHypervisor(hv_name)
2440
          utils.ForceDictType(hv_params, constants.HVS_PARAMETER_TYPES)
2441
          hv_class.CheckParameterSyntax(hv_params)
2442
          _CheckHVParams(self, node_list, hv_name, hv_params)
2443

    
2444
    if self.op.os_hvp:
2445
      # no need to check any newly-enabled hypervisors, since the
2446
      # defaults have already been checked in the above code-block
2447
      for os_name, os_hvp in self.new_os_hvp.items():
2448
        for hv_name, hv_params in os_hvp.items():
2449
          utils.ForceDictType(hv_params, constants.HVS_PARAMETER_TYPES)
2450
          # we need to fill in the new os_hvp on top of the actual hv_p
2451
          cluster_defaults = self.new_hvparams.get(hv_name, {})
2452
          new_osp = objects.FillDict(cluster_defaults, hv_params)
2453
          hv_class = hypervisor.GetHypervisor(hv_name)
2454
          hv_class.CheckParameterSyntax(new_osp)
2455
          _CheckHVParams(self, node_list, hv_name, new_osp)
2456

    
2457

    
2458
  def Exec(self, feedback_fn):
2459
    """Change the parameters of the cluster.
2460

2461
    """
2462
    if self.op.vg_name is not None:
2463
      new_volume = self.op.vg_name
2464
      if not new_volume:
2465
        new_volume = None
2466
      if new_volume != self.cfg.GetVGName():
2467
        self.cfg.SetVGName(new_volume)
2468
      else:
2469
        feedback_fn("Cluster LVM configuration already in desired"
2470
                    " state, not changing")
2471
    if self.op.hvparams:
2472
      self.cluster.hvparams = self.new_hvparams
2473
    if self.op.os_hvp:
2474
      self.cluster.os_hvp = self.new_os_hvp
2475
    if self.op.enabled_hypervisors is not None:
2476
      self.cluster.hvparams = self.new_hvparams
2477
      self.cluster.enabled_hypervisors = self.op.enabled_hypervisors
2478
    if self.op.beparams:
2479
      self.cluster.beparams[constants.PP_DEFAULT] = self.new_beparams
2480
    if self.op.nicparams:
2481
      self.cluster.nicparams[constants.PP_DEFAULT] = self.new_nicparams
2482

    
2483
    if self.op.candidate_pool_size is not None:
2484
      self.cluster.candidate_pool_size = self.op.candidate_pool_size
2485
      # we need to update the pool size here, otherwise the save will fail
2486
      _AdjustCandidatePool(self, [])
2487

    
2488
    if self.op.maintain_node_health is not None:
2489
      self.cluster.maintain_node_health = self.op.maintain_node_health
2490

    
2491
    if self.op.add_uids is not None:
2492
      uidpool.AddToUidPool(self.cluster.uid_pool, self.op.add_uids)
2493

    
2494
    if self.op.remove_uids is not None:
2495
      uidpool.RemoveFromUidPool(self.cluster.uid_pool, self.op.remove_uids)
2496

    
2497
    if self.op.uid_pool is not None:
2498
      self.cluster.uid_pool = self.op.uid_pool
2499

    
2500
    self.cfg.Update(self.cluster, feedback_fn)
2501

    
2502

    
2503
def _RedistributeAncillaryFiles(lu, additional_nodes=None):
2504
  """Distribute additional files which are part of the cluster configuration.
2505

2506
  ConfigWriter takes care of distributing the config and ssconf files, but
2507
  there are more files which should be distributed to all nodes. This function
2508
  makes sure those are copied.
2509

2510
  @param lu: calling logical unit
2511
  @param additional_nodes: list of nodes not in the config to distribute to
2512

2513
  """
2514
  # 1. Gather target nodes
2515
  myself = lu.cfg.GetNodeInfo(lu.cfg.GetMasterNode())
2516
  dist_nodes = lu.cfg.GetOnlineNodeList()
2517
  if additional_nodes is not None:
2518
    dist_nodes.extend(additional_nodes)
2519
  if myself.name in dist_nodes:
2520
    dist_nodes.remove(myself.name)
2521

    
2522
  # 2. Gather files to distribute
2523
  dist_files = set([constants.ETC_HOSTS,
2524
                    constants.SSH_KNOWN_HOSTS_FILE,
2525
                    constants.RAPI_CERT_FILE,
2526
                    constants.RAPI_USERS_FILE,
2527
                    constants.CONFD_HMAC_KEY,
2528
                   ])
2529

    
2530
  enabled_hypervisors = lu.cfg.GetClusterInfo().enabled_hypervisors
2531
  for hv_name in enabled_hypervisors:
2532
    hv_class = hypervisor.GetHypervisor(hv_name)
2533
    dist_files.update(hv_class.GetAncillaryFiles())
2534

    
2535
  # 3. Perform the files upload
2536
  for fname in dist_files:
2537
    if os.path.exists(fname):
2538
      result = lu.rpc.call_upload_file(dist_nodes, fname)
2539
      for to_node, to_result in result.items():
2540
        msg = to_result.fail_msg
2541
        if msg:
2542
          msg = ("Copy of file %s to node %s failed: %s" %
2543
                 (fname, to_node, msg))
2544
          lu.proc.LogWarning(msg)
2545

    
2546

    
2547
class LURedistributeConfig(NoHooksLU):
2548
  """Force the redistribution of cluster configuration.
2549

2550
  This is a very simple LU.
2551

2552
  """
2553
  _OP_REQP = []
2554
  REQ_BGL = False
2555

    
2556
  def ExpandNames(self):
2557
    self.needed_locks = {
2558
      locking.LEVEL_NODE: locking.ALL_SET,
2559
    }
2560
    self.share_locks[locking.LEVEL_NODE] = 1
2561

    
2562
  def CheckPrereq(self):
2563
    """Check prerequisites.
2564

2565
    """
2566

    
2567
  def Exec(self, feedback_fn):
2568
    """Redistribute the configuration.
2569

2570
    """
2571
    self.cfg.Update(self.cfg.GetClusterInfo(), feedback_fn)
2572
    _RedistributeAncillaryFiles(self)
2573

    
2574

    
2575
def _WaitForSync(lu, instance, oneshot=False):
2576
  """Sleep and poll for an instance's disk to sync.
2577

2578
  """
2579
  if not instance.disks:
2580
    return True
2581

    
2582
  if not oneshot:
2583
    lu.proc.LogInfo("Waiting for instance %s to sync disks." % instance.name)
2584

    
2585
  node = instance.primary_node
2586

    
2587
  for dev in instance.disks:
2588
    lu.cfg.SetDiskID(dev, node)
2589

    
2590
  # TODO: Convert to utils.Retry
2591

    
2592
  retries = 0
2593
  degr_retries = 10 # in seconds, as we sleep 1 second each time
2594
  while True:
2595
    max_time = 0
2596
    done = True
2597
    cumul_degraded = False
2598
    rstats = lu.rpc.call_blockdev_getmirrorstatus(node, instance.disks)
2599
    msg = rstats.fail_msg
2600
    if msg:
2601
      lu.LogWarning("Can't get any data from node %s: %s", node, msg)
2602
      retries += 1
2603
      if retries >= 10:
2604
        raise errors.RemoteError("Can't contact node %s for mirror data,"
2605
                                 " aborting." % node)
2606
      time.sleep(6)
2607
      continue
2608
    rstats = rstats.payload
2609
    retries = 0
2610
    for i, mstat in enumerate(rstats):
2611
      if mstat is None:
2612
        lu.LogWarning("Can't compute data for node %s/%s",
2613
                           node, instance.disks[i].iv_name)
2614
        continue
2615

    
2616
      cumul_degraded = (cumul_degraded or
2617
                        (mstat.is_degraded and mstat.sync_percent is None))
2618
      if mstat.sync_percent is not None:
2619
        done = False
2620
        if mstat.estimated_time is not None:
2621
          rem_time = "%d estimated seconds remaining" % mstat.estimated_time
2622
          max_time = mstat.estimated_time
2623
        else:
2624
          rem_time = "no time estimate"
2625
        lu.proc.LogInfo("- device %s: %5.2f%% done, %s" %
2626
                        (instance.disks[i].iv_name, mstat.sync_percent,
2627
                         rem_time))
2628

    
2629
    # if we're done but degraded, let's do a few small retries, to
2630
    # make sure we see a stable and not transient situation; therefore
2631
    # we force restart of the loop
2632
    if (done or oneshot) and cumul_degraded and degr_retries > 0:
2633
      logging.info("Degraded disks found, %d retries left", degr_retries)
2634
      degr_retries -= 1
2635
      time.sleep(1)
2636
      continue
2637

    
2638
    if done or oneshot:
2639
      break
2640

    
2641
    time.sleep(min(60, max_time))
2642

    
2643
  if done:
2644
    lu.proc.LogInfo("Instance %s's disks are in sync." % instance.name)
2645
  return not cumul_degraded
2646

    
2647

    
2648
def _CheckDiskConsistency(lu, dev, node, on_primary, ldisk=False):
2649
  """Check that mirrors are not degraded.
2650

2651
  The ldisk parameter, if True, will change the test from the
2652
  is_degraded attribute (which represents overall non-ok status for
2653
  the device(s)) to the ldisk (representing the local storage status).
2654

2655
  """
2656
  lu.cfg.SetDiskID(dev, node)
2657

    
2658
  result = True
2659

    
2660
  if on_primary or dev.AssembleOnSecondary():
2661
    rstats = lu.rpc.call_blockdev_find(node, dev)
2662
    msg = rstats.fail_msg
2663
    if msg:
2664
      lu.LogWarning("Can't find disk on node %s: %s", node, msg)
2665
      result = False
2666
    elif not rstats.payload:
2667
      lu.LogWarning("Can't find disk on node %s", node)
2668
      result = False
2669
    else:
2670
      if ldisk:
2671
        result = result and rstats.payload.ldisk_status == constants.LDS_OKAY
2672
      else:
2673
        result = result and not rstats.payload.is_degraded
2674

    
2675
  if dev.children:
2676
    for child in dev.children:
2677
      result = result and _CheckDiskConsistency(lu, child, node, on_primary)
2678

    
2679
  return result
2680

    
2681

    
2682
class LUDiagnoseOS(NoHooksLU):
2683
  """Logical unit for OS diagnose/query.
2684

2685
  """
2686
  _OP_REQP = ["output_fields", "names"]
2687
  REQ_BGL = False
2688
  _FIELDS_STATIC = utils.FieldSet()
2689
  _FIELDS_DYNAMIC = utils.FieldSet("name", "valid", "node_status", "variants")
2690
  # Fields that need calculation of global os validity
2691
  _FIELDS_NEEDVALID = frozenset(["valid", "variants"])
2692

    
2693
  def ExpandNames(self):
2694
    if self.op.names:
2695
      raise errors.OpPrereqError("Selective OS query not supported",
2696
                                 errors.ECODE_INVAL)
2697

    
2698
    _CheckOutputFields(static=self._FIELDS_STATIC,
2699
                       dynamic=self._FIELDS_DYNAMIC,
2700
                       selected=self.op.output_fields)
2701

    
2702
    # Lock all nodes, in shared mode
2703
    # Temporary removal of locks, should be reverted later
2704
    # TODO: reintroduce locks when they are lighter-weight
2705
    self.needed_locks = {}
2706
    #self.share_locks[locking.LEVEL_NODE] = 1
2707
    #self.needed_locks[locking.LEVEL_NODE] = locking.ALL_SET
2708

    
2709
  def CheckPrereq(self):
2710
    """Check prerequisites.
2711

2712
    """
2713

    
2714
  @staticmethod
2715
  def _DiagnoseByOS(rlist):
2716
    """Remaps a per-node return list into an a per-os per-node dictionary
2717

2718
    @param rlist: a map with node names as keys and OS objects as values
2719

2720
    @rtype: dict
2721
    @return: a dictionary with osnames as keys and as value another map, with
2722
        nodes as keys and tuples of (path, status, diagnose) as values, eg::
2723

2724
          {"debian-etch": {"node1": [(/usr/lib/..., True, ""),
2725
                                     (/srv/..., False, "invalid api")],
2726
                           "node2": [(/srv/..., True, "")]}
2727
          }
2728

2729
    """
2730
    all_os = {}
2731
    # we build here the list of nodes that didn't fail the RPC (at RPC
2732
    # level), so that nodes with a non-responding node daemon don't
2733
    # make all OSes invalid
2734
    good_nodes = [node_name for node_name in rlist
2735
                  if not rlist[node_name].fail_msg]
2736
    for node_name, nr in rlist.items():
2737
      if nr.fail_msg or not nr.payload:
2738
        continue
2739
      for name, path, status, diagnose, variants in nr.payload:
2740
        if name not in all_os:
2741
          # build a list of nodes for this os containing empty lists
2742
          # for each node in node_list
2743
          all_os[name] = {}
2744
          for nname in good_nodes:
2745
            all_os[name][nname] = []
2746
        all_os[name][node_name].append((path, status, diagnose, variants))
2747
    return all_os
2748

    
2749
  def Exec(self, feedback_fn):
2750
    """Compute the list of OSes.
2751

2752
    """
2753
    valid_nodes = [node for node in self.cfg.GetOnlineNodeList()]
2754
    node_data = self.rpc.call_os_diagnose(valid_nodes)
2755
    pol = self._DiagnoseByOS(node_data)
2756
    output = []
2757
    calc_valid = self._FIELDS_NEEDVALID.intersection(self.op.output_fields)
2758
    calc_variants = "variants" in self.op.output_fields
2759

    
2760
    for os_name, os_data in pol.items():
2761
      row = []
2762
      if calc_valid:
2763
        valid = True
2764
        variants = None
2765
        for osl in os_data.values():
2766
          valid = valid and osl and osl[0][1]
2767
          if not valid:
2768
            variants = None
2769
            break
2770
          if calc_variants:
2771
            node_variants = osl[0][3]
2772
            if variants is None:
2773
              variants = node_variants
2774
            else:
2775
              variants = [v for v in variants if v in node_variants]
2776

    
2777
      for field in self.op.output_fields:
2778
        if field == "name":
2779
          val = os_name
2780
        elif field == "valid":
2781
          val = valid
2782
        elif field == "node_status":
2783
          # this is just a copy of the dict
2784
          val = {}
2785
          for node_name, nos_list in os_data.items():
2786
            val[node_name] = nos_list
2787
        elif field == "variants":
2788
          val =  variants
2789
        else:
2790
          raise errors.ParameterError(field)
2791
        row.append(val)
2792
      output.append(row)
2793

    
2794
    return output
2795

    
2796

    
2797
class LURemoveNode(LogicalUnit):
2798
  """Logical unit for removing a node.
2799

2800
  """
2801
  HPATH = "node-remove"
2802
  HTYPE = constants.HTYPE_NODE
2803
  _OP_REQP = ["node_name"]
2804

    
2805
  def BuildHooksEnv(self):
2806
    """Build hooks env.
2807

2808
    This doesn't run on the target node in the pre phase as a failed
2809
    node would then be impossible to remove.
2810

2811
    """
2812
    env = {
2813
      "OP_TARGET": self.op.node_name,
2814
      "NODE_NAME": self.op.node_name,
2815
      }
2816
    all_nodes = self.cfg.GetNodeList()
2817
    try:
2818
      all_nodes.remove(self.op.node_name)
2819
    except ValueError:
2820
      logging.warning("Node %s which is about to be removed not found"
2821
                      " in the all nodes list", self.op.node_name)
2822
    return env, all_nodes, all_nodes
2823

    
2824
  def CheckPrereq(self):
2825
    """Check prerequisites.
2826

2827
    This checks:
2828
     - the node exists in the configuration
2829
     - it does not have primary or secondary instances
2830
     - it's not the master
2831

2832
    Any errors are signaled by raising errors.OpPrereqError.
2833

2834
    """
2835
    self.op.node_name = _ExpandNodeName(self.cfg, self.op.node_name)
2836
    node = self.cfg.GetNodeInfo(self.op.node_name)
2837
    assert node is not None
2838

    
2839
    instance_list = self.cfg.GetInstanceList()
2840

    
2841
    masternode = self.cfg.GetMasterNode()
2842
    if node.name == masternode:
2843
      raise errors.OpPrereqError("Node is the master node,"
2844
                                 " you need to failover first.",
2845
                                 errors.ECODE_INVAL)
2846

    
2847
    for instance_name in instance_list:
2848
      instance = self.cfg.GetInstanceInfo(instance_name)
2849
      if node.name in instance.all_nodes:
2850
        raise errors.OpPrereqError("Instance %s is still running on the node,"
2851
                                   " please remove first." % instance_name,
2852
                                   errors.ECODE_INVAL)
2853
    self.op.node_name = node.name
2854
    self.node = node
2855

    
2856
  def Exec(self, feedback_fn):
2857
    """Removes the node from the cluster.
2858

2859
    """
2860
    node = self.node
2861
    logging.info("Stopping the node daemon and removing configs from node %s",
2862
                 node.name)
2863

    
2864
    modify_ssh_setup = self.cfg.GetClusterInfo().modify_ssh_setup
2865

    
2866
    # Promote nodes to master candidate as needed
2867
    _AdjustCandidatePool(self, exceptions=[node.name])
2868
    self.context.RemoveNode(node.name)
2869

    
2870
    # Run post hooks on the node before it's removed
2871
    hm = self.proc.hmclass(self.rpc.call_hooks_runner, self)
2872
    try:
2873
      hm.RunPhase(constants.HOOKS_PHASE_POST, [node.name])
2874
    except:
2875
      # pylint: disable-msg=W0702
2876
      self.LogWarning("Errors occurred running hooks on %s" % node.name)
2877

    
2878
    result = self.rpc.call_node_leave_cluster(node.name, modify_ssh_setup)
2879
    msg = result.fail_msg
2880
    if msg:
2881
      self.LogWarning("Errors encountered on the remote node while leaving"
2882
                      " the cluster: %s", msg)
2883

    
2884

    
2885
class LUQueryNodes(NoHooksLU):
2886
  """Logical unit for querying nodes.
2887

2888
  """
2889
  # pylint: disable-msg=W0142
2890
  _OP_REQP = ["output_fields", "names", "use_locking"]
2891
  REQ_BGL = False
2892

    
2893
  _SIMPLE_FIELDS = ["name", "serial_no", "ctime", "mtime", "uuid",
2894
                    "master_candidate", "offline", "drained"]
2895

    
2896
  _FIELDS_DYNAMIC = utils.FieldSet(
2897
    "dtotal", "dfree",
2898
    "mtotal", "mnode", "mfree",
2899
    "bootid",
2900
    "ctotal", "cnodes", "csockets",
2901
    )
2902

    
2903
  _FIELDS_STATIC = utils.FieldSet(*[
2904
    "pinst_cnt", "sinst_cnt",
2905
    "pinst_list", "sinst_list",
2906
    "pip", "sip", "tags",
2907
    "master",
2908
    "role"] + _SIMPLE_FIELDS
2909
    )
2910

    
2911
  def ExpandNames(self):
2912
    _CheckOutputFields(static=self._FIELDS_STATIC,
2913
                       dynamic=self._FIELDS_DYNAMIC,
2914
                       selected=self.op.output_fields)
2915

    
2916
    self.needed_locks = {}
2917
    self.share_locks[locking.LEVEL_NODE] = 1
2918

    
2919
    if self.op.names:
2920
      self.wanted = _GetWantedNodes(self, self.op.names)
2921
    else:
2922
      self.wanted = locking.ALL_SET
2923

    
2924
    self.do_node_query = self._FIELDS_STATIC.NonMatching(self.op.output_fields)
2925
    self.do_locking = self.do_node_query and self.op.use_locking
2926
    if self.do_locking:
2927
      # if we don't request only static fields, we need to lock the nodes
2928
      self.needed_locks[locking.LEVEL_NODE] = self.wanted
2929

    
2930
  def CheckPrereq(self):
2931
    """Check prerequisites.
2932

2933
    """
2934
    # The validation of the node list is done in the _GetWantedNodes,
2935
    # if non empty, and if empty, there's no validation to do
2936
    pass
2937

    
2938
  def Exec(self, feedback_fn):
2939
    """Computes the list of nodes and their attributes.
2940

2941
    """
2942
    all_info = self.cfg.GetAllNodesInfo()
2943
    if self.do_locking:
2944
      nodenames = self.acquired_locks[locking.LEVEL_NODE]
2945
    elif self.wanted != locking.ALL_SET:
2946
      nodenames = self.wanted
2947
      missing = set(nodenames).difference(all_info.keys())
2948
      if missing:
2949
        raise errors.OpExecError(
2950
          "Some nodes were removed before retrieving their data: %s" % missing)
2951
    else:
2952
      nodenames = all_info.keys()
2953

    
2954
    nodenames = utils.NiceSort(nodenames)
2955
    nodelist = [all_info[name] for name in nodenames]
2956

    
2957
    # begin data gathering
2958

    
2959
    if self.do_node_query:
2960
      live_data = {}
2961
      node_data = self.rpc.call_node_info(nodenames, self.cfg.GetVGName(),
2962
                                          self.cfg.GetHypervisorType())
2963
      for name in nodenames:
2964
        nodeinfo = node_data[name]
2965
        if not nodeinfo.fail_msg and nodeinfo.payload:
2966
          nodeinfo = nodeinfo.payload
2967
          fn = utils.TryConvert
2968
          live_data[name] = {
2969
            "mtotal": fn(int, nodeinfo.get('memory_total', None)),
2970
            "mnode": fn(int, nodeinfo.get('memory_dom0', None)),
2971
            "mfree": fn(int, nodeinfo.get('memory_free', None)),
2972
            "dtotal": fn(int, nodeinfo.get('vg_size', None)),
2973
            "dfree": fn(int, nodeinfo.get('vg_free', None)),
2974
            "ctotal": fn(int, nodeinfo.get('cpu_total', None)),
2975
            "bootid": nodeinfo.get('bootid', None),
2976
            "cnodes": fn(int, nodeinfo.get('cpu_nodes', None)),
2977
            "csockets": fn(int, nodeinfo.get('cpu_sockets', None)),
2978
            }
2979
        else:
2980
          live_data[name] = {}
2981
    else:
2982
      live_data = dict.fromkeys(nodenames, {})
2983

    
2984
    node_to_primary = dict([(name, set()) for name in nodenames])
2985
    node_to_secondary = dict([(name, set()) for name in nodenames])
2986

    
2987
    inst_fields = frozenset(("pinst_cnt", "pinst_list",
2988
                             "sinst_cnt", "sinst_list"))
2989
    if inst_fields & frozenset(self.op.output_fields):
2990
      inst_data = self.cfg.GetAllInstancesInfo()
2991

    
2992
      for inst in inst_data.values():
2993
        if inst.primary_node in node_to_primary:
2994
          node_to_primary[inst.primary_node].add(inst.name)
2995
        for secnode in inst.secondary_nodes:
2996
          if secnode in node_to_secondary:
2997
            node_to_secondary[secnode].add(inst.name)
2998

    
2999
    master_node = self.cfg.GetMasterNode()
3000

    
3001
    # end data gathering
3002

    
3003
    output = []
3004
    for node in nodelist:
3005
      node_output = []
3006
      for field in self.op.output_fields:
3007
        if field in self._SIMPLE_FIELDS:
3008
          val = getattr(node, field)
3009
        elif field == "pinst_list":
3010
          val = list(node_to_primary[node.name])
3011
        elif field == "sinst_list":
3012
          val = list(node_to_secondary[node.name])
3013
        elif field == "pinst_cnt":
3014
          val = len(node_to_primary[node.name])
3015
        elif field == "sinst_cnt":
3016
          val = len(node_to_secondary[node.name])
3017
        elif field == "pip":
3018
          val = node.primary_ip
3019
        elif field == "sip":
3020
          val = node.secondary_ip
3021
        elif field == "tags":
3022
          val = list(node.GetTags())
3023
        elif field == "master":
3024
          val = node.name == master_node
3025
        elif self._FIELDS_DYNAMIC.Matches(field):
3026
          val = live_data[node.name].get(field, None)
3027
        elif field == "role":
3028
          if node.name == master_node:
3029
            val = "M"
3030
          elif node.master_candidate:
3031
            val = "C"
3032
          elif node.drained:
3033
            val = "D"
3034
          elif node.offline:
3035
            val = "O"
3036
          else:
3037
            val = "R"
3038
        else:
3039
          raise errors.ParameterError(field)
3040
        node_output.append(val)
3041
      output.append(node_output)
3042

    
3043
    return output
3044

    
3045

    
3046
class LUQueryNodeVolumes(NoHooksLU):
3047
  """Logical unit for getting volumes on node(s).
3048

3049
  """
3050
  _OP_REQP = ["nodes", "output_fields"]
3051
  REQ_BGL = False
3052
  _FIELDS_DYNAMIC = utils.FieldSet("phys", "vg", "name", "size", "instance")
3053
  _FIELDS_STATIC = utils.FieldSet("node")
3054

    
3055
  def ExpandNames(self):
3056
    _CheckOutputFields(static=self._FIELDS_STATIC,
3057
                       dynamic=self._FIELDS_DYNAMIC,
3058
                       selected=self.op.output_fields)
3059

    
3060
    self.needed_locks = {}
3061
    self.share_locks[locking.LEVEL_NODE] = 1
3062
    if not self.op.nodes:
3063
      self.needed_locks[locking.LEVEL_NODE] = locking.ALL_SET
3064
    else:
3065
      self.needed_locks[locking.LEVEL_NODE] = \
3066
        _GetWantedNodes(self, self.op.nodes)
3067

    
3068
  def CheckPrereq(self):
3069
    """Check prerequisites.
3070

3071
    This checks that the fields required are valid output fields.
3072

3073
    """
3074
    self.nodes = self.acquired_locks[locking.LEVEL_NODE]
3075

    
3076
  def Exec(self, feedback_fn):
3077
    """Computes the list of nodes and their attributes.
3078

3079
    """
3080
    nodenames = self.nodes
3081
    volumes = self.rpc.call_node_volumes(nodenames)
3082

    
3083
    ilist = [self.cfg.GetInstanceInfo(iname) for iname
3084
             in self.cfg.GetInstanceList()]
3085

    
3086
    lv_by_node = dict([(inst, inst.MapLVsByNode()) for inst in ilist])
3087

    
3088
    output = []
3089
    for node in nodenames:
3090
      nresult = volumes[node]
3091
      if nresult.offline:
3092
        continue
3093
      msg = nresult.fail_msg
3094
      if msg:
3095
        self.LogWarning("Can't compute volume data on node %s: %s", node, msg)
3096
        continue
3097

    
3098
      node_vols = nresult.payload[:]
3099
      node_vols.sort(key=lambda vol: vol['dev'])
3100

    
3101
      for vol in node_vols:
3102
        node_output = []
3103
        for field in self.op.output_fields:
3104
          if field == "node":
3105
            val = node
3106
          elif field == "phys":
3107
            val = vol['dev']
3108
          elif field == "vg":
3109
            val = vol['vg']
3110
          elif field == "name":
3111
            val = vol['name']
3112
          elif field == "size":
3113
            val = int(float(vol['size']))
3114
          elif field == "instance":
3115
            for inst in ilist:
3116
              if node not in lv_by_node[inst]:
3117
                continue
3118
              if vol['name'] in lv_by_node[inst][node]:
3119
                val = inst.name
3120
                break
3121
            else:
3122
              val = '-'
3123
          else:
3124
            raise errors.ParameterError(field)
3125
          node_output.append(str(val))
3126

    
3127
        output.append(node_output)
3128

    
3129
    return output
3130

    
3131

    
3132
class LUQueryNodeStorage(NoHooksLU):
3133
  """Logical unit for getting information on storage units on node(s).
3134

3135
  """
3136
  _OP_REQP = ["nodes", "storage_type", "output_fields"]
3137
  REQ_BGL = False
3138
  _FIELDS_STATIC = utils.FieldSet(constants.SF_NODE)
3139

    
3140
  def CheckArguments(self):
3141
    _CheckStorageType(self.op.storage_type)
3142

    
3143
    _CheckOutputFields(static=self._FIELDS_STATIC,
3144
                       dynamic=utils.FieldSet(*constants.VALID_STORAGE_FIELDS),
3145
                       selected=self.op.output_fields)
3146

    
3147
  def ExpandNames(self):
3148
    self.needed_locks = {}
3149
    self.share_locks[locking.LEVEL_NODE] = 1
3150

    
3151
    if self.op.nodes:
3152
      self.needed_locks[locking.LEVEL_NODE] = \
3153
        _GetWantedNodes(self, self.op.nodes)
3154
    else:
3155
      self.needed_locks[locking.LEVEL_NODE] = locking.ALL_SET
3156

    
3157
  def CheckPrereq(self):
3158
    """Check prerequisites.
3159

3160
    This checks that the fields required are valid output fields.
3161

3162
    """
3163
    self.op.name = getattr(self.op, "name", None)
3164

    
3165
    self.nodes = self.acquired_locks[locking.LEVEL_NODE]
3166

    
3167
  def Exec(self, feedback_fn):
3168
    """Computes the list of nodes and their attributes.
3169

3170
    """
3171
    # Always get name to sort by
3172
    if constants.SF_NAME in self.op.output_fields:
3173
      fields = self.op.output_fields[:]
3174
    else:
3175
      fields = [constants.SF_NAME] + self.op.output_fields
3176

    
3177
    # Never ask for node or type as it's only known to the LU
3178
    for extra in [constants.SF_NODE, constants.SF_TYPE]:
3179
      while extra in fields:
3180
        fields.remove(extra)
3181

    
3182
    field_idx = dict([(name, idx) for (idx, name) in enumerate(fields)])
3183
    name_idx = field_idx[constants.SF_NAME]
3184

    
3185
    st_args = _GetStorageTypeArgs(self.cfg, self.op.storage_type)
3186
    data = self.rpc.call_storage_list(self.nodes,
3187
                                      self.op.storage_type, st_args,
3188
                                      self.op.name, fields)
3189

    
3190
    result = []
3191

    
3192
    for node in utils.NiceSort(self.nodes):
3193
      nresult = data[node]
3194
      if nresult.offline:
3195
        continue
3196

    
3197
      msg = nresult.fail_msg
3198
      if msg:
3199
        self.LogWarning("Can't get storage data from node %s: %s", node, msg)
3200
        continue
3201

    
3202
      rows = dict([(row[name_idx], row) for row in nresult.payload])
3203

    
3204
      for name in utils.NiceSort(rows.keys()):
3205
        row = rows[name]
3206

    
3207
        out = []
3208

    
3209
        for field in self.op.output_fields:
3210
          if field == constants.SF_NODE:
3211
            val = node
3212
          elif field == constants.SF_TYPE:
3213
            val = self.op.storage_type
3214
          elif field in field_idx:
3215
            val = row[field_idx[field]]
3216
          else:
3217
            raise errors.ParameterError(field)
3218

    
3219
          out.append(val)
3220

    
3221
        result.append(out)
3222

    
3223
    return result
3224

    
3225

    
3226
class LUModifyNodeStorage(NoHooksLU):
3227
  """Logical unit for modifying a storage volume on a node.
3228

3229
  """
3230
  _OP_REQP = ["node_name", "storage_type", "name", "changes"]
3231
  REQ_BGL = False
3232

    
3233
  def CheckArguments(self):
3234
    self.opnode_name = _ExpandNodeName(self.cfg, self.op.node_name)
3235

    
3236
    _CheckStorageType(self.op.storage_type)
3237

    
3238
  def ExpandNames(self):
3239
    self.needed_locks = {
3240
      locking.LEVEL_NODE: self.op.node_name,
3241
      }
3242

    
3243
  def CheckPrereq(self):
3244
    """Check prerequisites.
3245

3246
    """
3247
    storage_type = self.op.storage_type
3248

    
3249
    try:
3250
      modifiable = constants.MODIFIABLE_STORAGE_FIELDS[storage_type]
3251
    except KeyError:
3252
      raise errors.OpPrereqError("Storage units of type '%s' can not be"
3253
                                 " modified" % storage_type,
3254
                                 errors.ECODE_INVAL)
3255

    
3256
    diff = set(self.op.changes.keys()) - modifiable
3257
    if diff:
3258
      raise errors.OpPrereqError("The following fields can not be modified for"
3259
                                 " storage units of type '%s': %r" %
3260
                                 (storage_type, list(diff)),
3261
                                 errors.ECODE_INVAL)
3262

    
3263
  def Exec(self, feedback_fn):
3264
    """Computes the list of nodes and their attributes.
3265

3266
    """
3267
    st_args = _GetStorageTypeArgs(self.cfg, self.op.storage_type)
3268
    result = self.rpc.call_storage_modify(self.op.node_name,
3269
                                          self.op.storage_type, st_args,
3270
                                          self.op.name, self.op.changes)
3271
    result.Raise("Failed to modify storage unit '%s' on %s" %
3272
                 (self.op.name, self.op.node_name))
3273

    
3274

    
3275
class LUAddNode(LogicalUnit):
3276
  """Logical unit for adding node to the cluster.
3277

3278
  """
3279
  HPATH = "node-add"
3280
  HTYPE = constants.HTYPE_NODE
3281
  _OP_REQP = ["node_name"]
3282

    
3283
  def CheckArguments(self):
3284
    # validate/normalize the node name
3285
    self.op.node_name = utils.HostInfo.NormalizeName(self.op.node_name)
3286

    
3287
  def BuildHooksEnv(self):
3288
    """Build hooks env.
3289

3290
    This will run on all nodes before, and on all nodes + the new node after.
3291

3292
    """
3293
    env = {
3294
      "OP_TARGET": self.op.node_name,
3295
      "NODE_NAME": self.op.node_name,
3296
      "NODE_PIP": self.op.primary_ip,
3297
      "NODE_SIP": self.op.secondary_ip,
3298
      }
3299
    nodes_0 = self.cfg.GetNodeList()
3300
    nodes_1 = nodes_0 + [self.op.node_name, ]
3301
    return env, nodes_0, nodes_1
3302

    
3303
  def CheckPrereq(self):
3304
    """Check prerequisites.
3305

3306
    This checks:
3307
     - the new node is not already in the config
3308
     - it is resolvable
3309
     - its parameters (single/dual homed) matches the cluster
3310

3311
    Any errors are signaled by raising errors.OpPrereqError.
3312

3313
    """
3314
    node_name = self.op.node_name
3315
    cfg = self.cfg
3316

    
3317
    dns_data = utils.GetHostInfo(node_name)
3318

    
3319
    node = dns_data.name
3320
    primary_ip = self.op.primary_ip = dns_data.ip
3321
    secondary_ip = getattr(self.op, "secondary_ip", None)
3322
    if secondary_ip is None:
3323
      secondary_ip = primary_ip
3324
    if not utils.IsValidIP(secondary_ip):
3325
      raise errors.OpPrereqError("Invalid secondary IP given",
3326
                                 errors.ECODE_INVAL)
3327
    self.op.secondary_ip = secondary_ip
3328

    
3329
    node_list = cfg.GetNodeList()
3330
    if not self.op.readd and node in node_list:
3331
      raise errors.OpPrereqError("Node %s is already in the configuration" %
3332
                                 node, errors.ECODE_EXISTS)
3333
    elif self.op.readd and node not in node_list:
3334
      raise errors.OpPrereqError("Node %s is not in the configuration" % node,
3335
                                 errors.ECODE_NOENT)
3336

    
3337
    self.changed_primary_ip = False
3338

    
3339
    for existing_node_name in node_list:
3340
      existing_node = cfg.GetNodeInfo(existing_node_name)
3341

    
3342
      if self.op.readd and node == existing_node_name:
3343
        if existing_node.secondary_ip != secondary_ip:
3344
          raise errors.OpPrereqError("Readded node doesn't have the same IP"
3345
                                     " address configuration as before",
3346
                                     errors.ECODE_INVAL)
3347
        if existing_node.primary_ip != primary_ip:
3348
          self.changed_primary_ip = True
3349

    
3350
        continue
3351

    
3352
      if (existing_node.primary_ip == primary_ip or
3353
          existing_node.secondary_ip == primary_ip or
3354
          existing_node.primary_ip == secondary_ip or
3355
          existing_node.secondary_ip == secondary_ip):
3356
        raise errors.OpPrereqError("New node ip address(es) conflict with"
3357
                                   " existing node %s" % existing_node.name,
3358
                                   errors.ECODE_NOTUNIQUE)
3359

    
3360
    # check that the type of the node (single versus dual homed) is the
3361
    # same as for the master
3362
    myself = cfg.GetNodeInfo(self.cfg.GetMasterNode())
3363
    master_singlehomed = myself.secondary_ip == myself.primary_ip
3364
    newbie_singlehomed = secondary_ip == primary_ip
3365
    if master_singlehomed != newbie_singlehomed:
3366
      if master_singlehomed:
3367
        raise errors.OpPrereqError("The master has no private ip but the"
3368
                                   " new node has one",
3369
                                   errors.ECODE_INVAL)
3370
      else:
3371
        raise errors.OpPrereqError("The master has a private ip but the"
3372
                                   " new node doesn't have one",
3373
                                   errors.ECODE_INVAL)
3374

    
3375
    # checks reachability
3376
    if not utils.TcpPing(primary_ip, constants.DEFAULT_NODED_PORT):
3377
      raise errors.OpPrereqError("Node not reachable by ping",
3378
                                 errors.ECODE_ENVIRON)
3379

    
3380
    if not newbie_singlehomed:
3381
      # check reachability from my secondary ip to newbie's secondary ip
3382
      if not utils.TcpPing(secondary_ip, constants.DEFAULT_NODED_PORT,
3383
                           source=myself.secondary_ip):
3384
        raise errors.OpPrereqError("Node secondary ip not reachable by TCP"
3385
                                   " based ping to noded port",
3386
                                   errors.ECODE_ENVIRON)
3387

    
3388
    if self.op.readd:
3389
      exceptions = [node]
3390
    else:
3391
      exceptions = []
3392

    
3393
    self.master_candidate = _DecideSelfPromotion(self, exceptions=exceptions)
3394

    
3395
    if self.op.readd:
3396
      self.new_node = self.cfg.GetNodeInfo(node)
3397
      assert self.new_node is not None, "Can't retrieve locked node %s" % node
3398
    else:
3399
      self.new_node = objects.Node(name=node,
3400
                                   primary_ip=primary_ip,
3401
                                   secondary_ip=secondary_ip,
3402
                                   master_candidate=self.master_candidate,
3403
                                   offline=False, drained=False)
3404

    
3405
  def Exec(self, feedback_fn):
3406
    """Adds the new node to the cluster.
3407

3408
    """
3409
    new_node = self.new_node
3410
    node = new_node.name
3411

    
3412
    # for re-adds, reset the offline/drained/master-candidate flags;
3413
    # we need to reset here, otherwise offline would prevent RPC calls
3414
    # later in the procedure; this also means that if the re-add
3415
    # fails, we are left with a non-offlined, broken node
3416
    if self.op.readd:
3417
      new_node.drained = new_node.offline = False # pylint: disable-msg=W0201
3418
      self.LogInfo("Readding a node, the offline/drained flags were reset")
3419
      # if we demote the node, we do cleanup later in the procedure
3420
      new_node.master_candidate = self.master_candidate
3421
      if self.changed_primary_ip:
3422
        new_node.primary_ip = self.op.primary_ip
3423

    
3424
    # notify the user about any possible mc promotion
3425
    if new_node.master_candidate:
3426
      self.LogInfo("Node will be a master candidate")
3427

    
3428
    # check connectivity
3429
    result = self.rpc.call_version([node])[node]
3430
    result.Raise("Can't get version information from node %s" % node)
3431
    if constants.PROTOCOL_VERSION == result.payload:
3432
      logging.info("Communication to node %s fine, sw version %s match",
3433
                   node, result.payload)
3434
    else:
3435
      raise errors.OpExecError("Version mismatch master version %s,"
3436
                               " node version %s" %
3437
                               (constants.PROTOCOL_VERSION, result.payload))
3438

    
3439
    # setup ssh on node
3440
    if self.cfg.GetClusterInfo().modify_ssh_setup:
3441
      logging.info("Copy ssh key to node %s", node)
3442
      priv_key, pub_key, _ = ssh.GetUserFiles(constants.GANETI_RUNAS)
3443
      keyarray = []
3444
      keyfiles = [constants.SSH_HOST_DSA_PRIV, constants.SSH_HOST_DSA_PUB,
3445
                  constants.SSH_HOST_RSA_PRIV, constants.SSH_HOST_RSA_PUB,
3446
                  priv_key, pub_key]
3447

    
3448
      for i in keyfiles:
3449
        keyarray.append(utils.ReadFile(i))
3450

    
3451
      result = self.rpc.call_node_add(node, keyarray[0], keyarray[1],
3452
                                      keyarray[2], keyarray[3], keyarray[4],
3453
                                      keyarray[5])
3454
      result.Raise("Cannot transfer ssh keys to the new node")
3455

    
3456
    # Add node to our /etc/hosts, and add key to known_hosts
3457
    if self.cfg.GetClusterInfo().modify_etc_hosts:
3458
      utils.AddHostToEtcHosts(new_node.name)
3459

    
3460
    if new_node.secondary_ip != new_node.primary_ip:
3461
      result = self.rpc.call_node_has_ip_address(new_node.name,
3462
                                                 new_node.secondary_ip)
3463
      result.Raise("Failure checking secondary ip on node %s" % new_node.name,
3464
                   prereq=True, ecode=errors.ECODE_ENVIRON)
3465
      if not result.payload:
3466
        raise errors.OpExecError("Node claims it doesn't have the secondary ip"
3467
                                 " you gave (%s). Please fix and re-run this"
3468
                                 " command." % new_node.secondary_ip)
3469

    
3470
    node_verify_list = [self.cfg.GetMasterNode()]
3471
    node_verify_param = {
3472
      constants.NV_NODELIST: [node],
3473
      # TODO: do a node-net-test as well?
3474
    }
3475

    
3476
    result = self.rpc.call_node_verify(node_verify_list, node_verify_param,
3477
                                       self.cfg.GetClusterName())
3478
    for verifier in node_verify_list:
3479
      result[verifier].Raise("Cannot communicate with node %s" % verifier)
3480
      nl_payload = result[verifier].payload[constants.NV_NODELIST]
3481
      if nl_payload:
3482
        for failed in nl_payload:
3483
          feedback_fn("ssh/hostname verification failed"
3484
                      " (checking from %s): %s" %
3485
                      (verifier, nl_payload[failed]))
3486
        raise errors.OpExecError("ssh/hostname verification failed.")
3487

    
3488
    if self.op.readd:
3489
      _RedistributeAncillaryFiles(self)
3490
      self.context.ReaddNode(new_node)
3491
      # make sure we redistribute the config
3492
      self.cfg.Update(new_node, feedback_fn)
3493
      # and make sure the new node will not have old files around
3494
      if not new_node.master_candidate:
3495
        result = self.rpc.call_node_demote_from_mc(new_node.name)
3496
        msg = result.fail_msg
3497
        if msg:
3498
          self.LogWarning("Node failed to demote itself from master"
3499
                          " candidate status: %s" % msg)
3500
    else:
3501
      _RedistributeAncillaryFiles(self, additional_nodes=[node])
3502
      self.context.AddNode(new_node, self.proc.GetECId())
3503

    
3504

    
3505
class LUSetNodeParams(LogicalUnit):
3506
  """Modifies the parameters of a node.
3507

3508
  """
3509
  HPATH = "node-modify"
3510
  HTYPE = constants.HTYPE_NODE
3511
  _OP_REQP = ["node_name"]
3512
  REQ_BGL = False
3513

    
3514
  def CheckArguments(self):
3515
    self.op.node_name = _ExpandNodeName(self.cfg, self.op.node_name)
3516
    _CheckBooleanOpField(self.op, 'master_candidate')
3517
    _CheckBooleanOpField(self.op, 'offline')
3518
    _CheckBooleanOpField(self.op, 'drained')
3519
    _CheckBooleanOpField(self.op, 'auto_promote')
3520
    all_mods = [self.op.offline, self.op.master_candidate, self.op.drained]
3521
    if all_mods.count(None) == 3:
3522
      raise errors.OpPrereqError("Please pass at least one modification",
3523
                                 errors.ECODE_INVAL)
3524
    if all_mods.count(True) > 1:
3525
      raise errors.OpPrereqError("Can't set the node into more than one"
3526
                                 " state at the same time",
3527
                                 errors.ECODE_INVAL)
3528

    
3529
    # Boolean value that tells us whether we're offlining or draining the node
3530
    self.offline_or_drain = (self.op.offline == True or
3531
                             self.op.drained == True)
3532
    self.deoffline_or_drain = (self.op.offline == False or
3533
                               self.op.drained == False)
3534
    self.might_demote = (self.op.master_candidate == False or
3535
                         self.offline_or_drain)
3536

    
3537
    self.lock_all = self.op.auto_promote and self.might_demote
3538

    
3539

    
3540
  def ExpandNames(self):
3541
    if self.lock_all:
3542
      self.needed_locks = {locking.LEVEL_NODE: locking.ALL_SET}
3543
    else:
3544
      self.needed_locks = {locking.LEVEL_NODE: self.op.node_name}
3545

    
3546
  def BuildHooksEnv(self):
3547
    """Build hooks env.
3548

3549
    This runs on the master node.
3550

3551
    """
3552
    env = {
3553
      "OP_TARGET": self.op.node_name,
3554
      "MASTER_CANDIDATE": str(self.op.master_candidate),
3555
      "OFFLINE": str(self.op.offline),
3556
      "DRAINED": str(self.op.drained),
3557
      }
3558
    nl = [self.cfg.GetMasterNode(),
3559
          self.op.node_name]
3560
    return env, nl, nl
3561

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

3565
    This only checks the instance list against the existing names.
3566

3567
    """
3568
    node = self.node = self.cfg.GetNodeInfo(self.op.node_name)
3569

    
3570
    if (self.op.master_candidate is not None or
3571
        self.op.drained is not None or
3572
        self.op.offline is not None):
3573
      # we can't change the master's node flags
3574
      if self.op.node_name == self.cfg.GetMasterNode():
3575
        raise errors.OpPrereqError("The master role can be changed"
3576
                                   " only via masterfailover",
3577
                                   errors.ECODE_INVAL)
3578

    
3579

    
3580
    if node.master_candidate and self.might_demote and not self.lock_all:
3581
      assert not self.op.auto_promote, "auto-promote set but lock_all not"
3582
      # check if after removing the current node, we're missing master
3583
      # candidates
3584
      (mc_remaining, mc_should, _) = \
3585
          self.cfg.GetMasterCandidateStats(exceptions=[node.name])
3586
      if mc_remaining < mc_should:
3587
        raise errors.OpPrereqError("Not enough master candidates, please"
3588
                                   " pass auto_promote to allow promotion",
3589
                                   errors.ECODE_INVAL)
3590

    
3591
    if (self.op.master_candidate == True and
3592
        ((node.offline and not self.op.offline == False) or
3593
         (node.drained and not self.op.drained == False))):
3594
      raise errors.OpPrereqError("Node '%s' is offline or drained, can't set"
3595
                                 " to master_candidate" % node.name,
3596
                                 errors.ECODE_INVAL)
3597

    
3598
    # If we're being deofflined/drained, we'll MC ourself if needed
3599
    if (self.deoffline_or_drain and not self.offline_or_drain and not
3600
        self.op.master_candidate == True and not node.master_candidate):
3601
      self.op.master_candidate = _DecideSelfPromotion(self)
3602
      if self.op.master_candidate:
3603
        self.LogInfo("Autopromoting node to master candidate")
3604

    
3605
    return
3606

    
3607
  def Exec(self, feedback_fn):
3608
    """Modifies a node.
3609

3610
    """
3611
    node = self.node
3612

    
3613
    result = []
3614
    changed_mc = False
3615

    
3616
    if self.op.offline is not None:
3617
      node.offline = self.op.offline
3618
      result.append(("offline", str(self.op.offline)))
3619
      if self.op.offline == True:
3620
        if node.master_candidate:
3621
          node.master_candidate = False
3622
          changed_mc = True
3623
          result.append(("master_candidate", "auto-demotion due to offline"))
3624
        if node.drained:
3625
          node.drained = False
3626
          result.append(("drained", "clear drained status due to offline"))
3627

    
3628
    if self.op.master_candidate is not None:
3629
      node.master_candidate = self.op.master_candidate
3630
      changed_mc = True
3631
      result.append(("master_candidate", str(self.op.master_candidate)))
3632
      if self.op.master_candidate == False:
3633
        rrc = self.rpc.call_node_demote_from_mc(node.name)
3634
        msg = rrc.fail_msg
3635
        if msg:
3636
          self.LogWarning("Node failed to demote itself: %s" % msg)
3637

    
3638
    if self.op.drained is not None:
3639
      node.drained = self.op.drained
3640
      result.append(("drained", str(self.op.drained)))
3641
      if self.op.drained == True:
3642
        if node.master_candidate:
3643
          node.master_candidate = False
3644
          changed_mc = True
3645
          result.append(("master_candidate", "auto-demotion due to drain"))
3646
          rrc = self.rpc.call_node_demote_from_mc(node.name)
3647
          msg = rrc.fail_msg
3648
          if msg:
3649
            self.LogWarning("Node failed to demote itself: %s" % msg)
3650
        if node.offline:
3651
          node.offline = False
3652
          result.append(("offline", "clear offline status due to drain"))
3653

    
3654
    # we locked all nodes, we adjust the CP before updating this node
3655
    if self.lock_all:
3656
      _AdjustCandidatePool(self, [node.name])
3657

    
3658
    # this will trigger configuration file update, if needed
3659
    self.cfg.Update(node, feedback_fn)
3660

    
3661
    # this will trigger job queue propagation or cleanup
3662
    if changed_mc:
3663
      self.context.ReaddNode(node)
3664

    
3665
    return result
3666

    
3667

    
3668
class LUPowercycleNode(NoHooksLU):
3669
  """Powercycles a node.
3670

3671
  """
3672
  _OP_REQP = ["node_name", "force"]
3673
  REQ_BGL = False
3674

    
3675
  def CheckArguments(self):
3676
    self.op.node_name = _ExpandNodeName(self.cfg, self.op.node_name)
3677
    if self.op.node_name == self.cfg.GetMasterNode() and not self.op.force:
3678
      raise errors.OpPrereqError("The node is the master and the force"
3679
                                 " parameter was not set",
3680
                                 errors.ECODE_INVAL)
3681

    
3682
  def ExpandNames(self):
3683
    """Locking for PowercycleNode.
3684

3685
    This is a last-resort option and shouldn't block on other
3686
    jobs. Therefore, we grab no locks.
3687

3688
    """
3689
    self.needed_locks = {}
3690

    
3691
  def CheckPrereq(self):
3692
    """Check prerequisites.
3693

3694
    This LU has no prereqs.
3695

3696
    """
3697
    pass
3698

    
3699
  def Exec(self, feedback_fn):
3700
    """Reboots a node.
3701

3702
    """
3703
    result = self.rpc.call_node_powercycle(self.op.node_name,
3704
                                           self.cfg.GetHypervisorType())
3705
    result.Raise("Failed to schedule the reboot")
3706
    return result.payload
3707

    
3708

    
3709
class LUQueryClusterInfo(NoHooksLU):
3710
  """Query cluster configuration.
3711

3712
  """
3713
  _OP_REQP = []
3714
  REQ_BGL = False
3715

    
3716
  def ExpandNames(self):
3717
    self.needed_locks = {}
3718

    
3719
  def CheckPrereq(self):
3720
    """No prerequsites needed for this LU.
3721

3722
    """
3723
    pass
3724

    
3725
  def Exec(self, feedback_fn):
3726
    """Return cluster config.
3727

3728
    """
3729
    cluster = self.cfg.GetClusterInfo()
3730
    os_hvp = {}
3731

    
3732
    # Filter just for enabled hypervisors
3733
    for os_name, hv_dict in cluster.os_hvp.items():
3734
      os_hvp[os_name] = {}
3735
      for hv_name, hv_params in hv_dict.items():
3736
        if hv_name in cluster.enabled_hypervisors:
3737
          os_hvp[os_name][hv_name] = hv_params
3738

    
3739
    result = {
3740
      "software_version": constants.RELEASE_VERSION,
3741
      "protocol_version": constants.PROTOCOL_VERSION,
3742
      "config_version": constants.CONFIG_VERSION,
3743
      "os_api_version": max(constants.OS_API_VERSIONS),
3744
      "export_version": constants.EXPORT_VERSION,
3745
      "architecture": (platform.architecture()[0], platform.machine()),
3746
      "name": cluster.cluster_name,
3747
      "master": cluster.master_node,
3748
      "default_hypervisor": cluster.enabled_hypervisors[0],
3749
      "enabled_hypervisors": cluster.enabled_hypervisors,
3750
      "hvparams": dict([(hypervisor_name, cluster.hvparams[hypervisor_name])
3751
                        for hypervisor_name in cluster.enabled_hypervisors]),
3752
      "os_hvp": os_hvp,
3753
      "beparams": cluster.beparams,
3754
      "nicparams": cluster.nicparams,
3755
      "candidate_pool_size": cluster.candidate_pool_size,
3756
      "master_netdev": cluster.master_netdev,
3757
      "volume_group_name": cluster.volume_group_name,
3758
      "file_storage_dir": cluster.file_storage_dir,
3759
      "maintain_node_health": cluster.maintain_node_health,
3760
      "ctime": cluster.ctime,
3761
      "mtime": cluster.mtime,
3762
      "uuid": cluster.uuid,
3763
      "tags": list(cluster.GetTags()),
3764
      "uid_pool": cluster.uid_pool,
3765
      }
3766

    
3767
    return result
3768

    
3769

    
3770
class LUQueryConfigValues(NoHooksLU):
3771
  """Return configuration values.
3772

3773
  """
3774
  _OP_REQP = []
3775
  REQ_BGL = False
3776
  _FIELDS_DYNAMIC = utils.FieldSet()
3777
  _FIELDS_STATIC = utils.FieldSet("cluster_name", "master_node", "drain_flag",
3778
                                  "watcher_pause")
3779

    
3780
  def ExpandNames(self):
3781
    self.needed_locks = {}
3782

    
3783
    _CheckOutputFields(static=self._FIELDS_STATIC,
3784
                       dynamic=self._FIELDS_DYNAMIC,
3785
                       selected=self.op.output_fields)
3786

    
3787
  def CheckPrereq(self):
3788
    """No prerequisites.
3789

3790
    """
3791
    pass
3792

    
3793
  def Exec(self, feedback_fn):
3794
    """Dump a representation of the cluster config to the standard output.
3795

3796
    """
3797
    values = []
3798
    for field in self.op.output_fields:
3799
      if field == "cluster_name":
3800
        entry = self.cfg.GetClusterName()
3801
      elif field == "master_node":
3802
        entry = self.cfg.GetMasterNode()
3803
      elif field == "drain_flag":
3804
        entry = os.path.exists(constants.JOB_QUEUE_DRAIN_FILE)
3805
      elif field == "watcher_pause":
3806
        entry = utils.ReadWatcherPauseFile(constants.WATCHER_PAUSEFILE)
3807
      else:
3808
        raise errors.ParameterError(field)
3809
      values.append(entry)
3810
    return values
3811

    
3812

    
3813
class LUActivateInstanceDisks(NoHooksLU):
3814
  """Bring up an instance's disks.
3815

3816
  """
3817
  _OP_REQP = ["instance_name"]
3818
  REQ_BGL = False
3819

    
3820
  def ExpandNames(self):
3821
    self._ExpandAndLockInstance()
3822
    self.needed_locks[locking.LEVEL_NODE] = []
3823
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
3824

    
3825
  def DeclareLocks(self, level):
3826
    if level == locking.LEVEL_NODE:
3827
      self._LockInstancesNodes()
3828

    
3829
  def CheckPrereq(self):
3830
    """Check prerequisites.
3831

3832
    This checks that the instance is in the cluster.
3833

3834
    """
3835
    self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
3836
    assert self.instance is not None, \
3837
      "Cannot retrieve locked instance %s" % self.op.instance_name
3838
    _CheckNodeOnline(self, self.instance.primary_node)
3839
    if not hasattr(self.op, "ignore_size"):
3840
      self.op.ignore_size = False
3841

    
3842
  def Exec(self, feedback_fn):
3843
    """Activate the disks.
3844

3845
    """
3846
    disks_ok, disks_info = \
3847
              _AssembleInstanceDisks(self, self.instance,
3848
                                     ignore_size=self.op.ignore_size)
3849
    if not disks_ok:
3850
      raise errors.OpExecError("Cannot activate block devices")
3851

    
3852
    return disks_info
3853

    
3854

    
3855
def _AssembleInstanceDisks(lu, instance, ignore_secondaries=False,
3856
                           ignore_size=False):
3857
  """Prepare the block devices for an instance.
3858

3859
  This sets up the block devices on all nodes.
3860

3861
  @type lu: L{LogicalUnit}
3862
  @param lu: the logical unit on whose behalf we execute
3863
  @type instance: L{objects.Instance}
3864
  @param instance: the instance for whose disks we assemble
3865
  @type ignore_secondaries: boolean
3866
  @param ignore_secondaries: if true, errors on secondary nodes
3867
      won't result in an error return from the function
3868
  @type ignore_size: boolean
3869
  @param ignore_size: if true, the current known size of the disk
3870
      will not be used during the disk activation, useful for cases
3871
      when the size is wrong
3872
  @return: False if the operation failed, otherwise a list of
3873
      (host, instance_visible_name, node_visible_name)
3874
      with the mapping from node devices to instance devices
3875

3876
  """
3877
  device_info = []
3878
  disks_ok = True
3879
  iname = instance.name
3880
  # With the two passes mechanism we try to reduce the window of
3881
  # opportunity for the race condition of switching DRBD to primary
3882
  # before handshaking occured, but we do not eliminate it
3883

    
3884
  # The proper fix would be to wait (with some limits) until the
3885
  # connection has been made and drbd transitions from WFConnection
3886
  # into any other network-connected state (Connected, SyncTarget,
3887
  # SyncSource, etc.)
3888

    
3889
  # 1st pass, assemble on all nodes in secondary mode
3890
  for inst_disk in instance.disks:
3891
    for node, node_disk in inst_disk.ComputeNodeTree(instance.primary_node):
3892
      if ignore_size:
3893
        node_disk = node_disk.Copy()
3894
        node_disk.UnsetSize()
3895
      lu.cfg.SetDiskID(node_disk, node)
3896
      result = lu.rpc.call_blockdev_assemble(node, node_disk, iname, False)
3897
      msg = result.fail_msg
3898
      if msg:
3899
        lu.proc.LogWarning("Could not prepare block device %s on node %s"
3900
                           " (is_primary=False, pass=1): %s",
3901
                           inst_disk.iv_name, node, msg)
3902
        if not ignore_secondaries:
3903
          disks_ok = False
3904

    
3905
  # FIXME: race condition on drbd migration to primary
3906

    
3907
  # 2nd pass, do only the primary node
3908
  for inst_disk in instance.disks:
3909
    dev_path = None
3910

    
3911
    for node, node_disk in inst_disk.ComputeNodeTree(instance.primary_node):
3912
      if node != instance.primary_node:
3913
        continue
3914
      if ignore_size:
3915
        node_disk = node_disk.Copy()
3916
        node_disk.UnsetSize()
3917
      lu.cfg.SetDiskID(node_disk, node)
3918
      result = lu.rpc.call_blockdev_assemble(node, node_disk, iname, True)
3919
      msg = result.fail_msg
3920
      if msg:
3921
        lu.proc.LogWarning("Could not prepare block device %s on node %s"
3922
                           " (is_primary=True, pass=2): %s",
3923
                           inst_disk.iv_name, node, msg)
3924
        disks_ok = False
3925
      else:
3926
        dev_path = result.payload
3927

    
3928
    device_info.append((instance.primary_node, inst_disk.iv_name, dev_path))
3929

    
3930
  # leave the disks configured for the primary node
3931
  # this is a workaround that would be fixed better by
3932
  # improving the logical/physical id handling
3933
  for disk in instance.disks:
3934
    lu.cfg.SetDiskID(disk, instance.primary_node)
3935

    
3936
  return disks_ok, device_info
3937

    
3938

    
3939
def _StartInstanceDisks(lu, instance, force):
3940
  """Start the disks of an instance.
3941

3942
  """
3943
  disks_ok, _ = _AssembleInstanceDisks(lu, instance,
3944
                                           ignore_secondaries=force)
3945
  if not disks_ok:
3946
    _ShutdownInstanceDisks(lu, instance)
3947
    if force is not None and not force:
3948
      lu.proc.LogWarning("", hint="If the message above refers to a"
3949
                         " secondary node,"
3950
                         " you can retry the operation using '--force'.")
3951
    raise errors.OpExecError("Disk consistency error")
3952

    
3953

    
3954
class LUDeactivateInstanceDisks(NoHooksLU):
3955
  """Shutdown an instance's disks.
3956

3957
  """
3958
  _OP_REQP = ["instance_name"]
3959
  REQ_BGL = False
3960

    
3961
  def ExpandNames(self):
3962
    self._ExpandAndLockInstance()
3963
    self.needed_locks[locking.LEVEL_NODE] = []
3964
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
3965

    
3966
  def DeclareLocks(self, level):
3967
    if level == locking.LEVEL_NODE:
3968
      self._LockInstancesNodes()
3969

    
3970
  def CheckPrereq(self):
3971
    """Check prerequisites.
3972

3973
    This checks that the instance is in the cluster.
3974

3975
    """
3976
    self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
3977
    assert self.instance is not None, \
3978
      "Cannot retrieve locked instance %s" % self.op.instance_name
3979

    
3980
  def Exec(self, feedback_fn):
3981
    """Deactivate the disks
3982

3983
    """
3984
    instance = self.instance
3985
    _SafeShutdownInstanceDisks(self, instance)
3986

    
3987

    
3988
def _SafeShutdownInstanceDisks(lu, instance):
3989
  """Shutdown block devices of an instance.
3990

3991
  This function checks if an instance is running, before calling
3992
  _ShutdownInstanceDisks.
3993

3994
  """
3995
  _CheckInstanceDown(lu, instance, "cannot shutdown disks")
3996
  _ShutdownInstanceDisks(lu, instance)
3997

    
3998

    
3999
def _ShutdownInstanceDisks(lu, instance, ignore_primary=False):
4000
  """Shutdown block devices of an instance.
4001

4002
  This does the shutdown on all nodes of the instance.
4003

4004
  If the ignore_primary is false, errors on the primary node are
4005
  ignored.
4006

4007
  """
4008
  all_result = True
4009
  for disk in instance.disks:
4010
    for node, top_disk in disk.ComputeNodeTree(instance.primary_node):
4011
      lu.cfg.SetDiskID(top_disk, node)
4012
      result = lu.rpc.call_blockdev_shutdown(node, top_disk)
4013
      msg = result.fail_msg
4014
      if msg:
4015
        lu.LogWarning("Could not shutdown block device %s on node %s: %s",
4016
                      disk.iv_name, node, msg)
4017
        if not ignore_primary or node != instance.primary_node:
4018
          all_result = False
4019
  return all_result
4020

    
4021

    
4022
def _CheckNodeFreeMemory(lu, node, reason, requested, hypervisor_name):
4023
  """Checks if a node has enough free memory.
4024

4025
  This function check if a given node has the needed amount of free
4026
  memory. In case the node has less memory or we cannot get the
4027
  information from the node, this function raise an OpPrereqError
4028
  exception.
4029

4030
  @type lu: C{LogicalUnit}
4031
  @param lu: a logical unit from which we get configuration data
4032
  @type node: C{str}
4033
  @param node: the node to check
4034
  @type reason: C{str}
4035
  @param reason: string to use in the error message
4036
  @type requested: C{int}
4037
  @param requested: the amount of memory in MiB to check for
4038
  @type hypervisor_name: C{str}
</