Statistics
| Branch: | Tag: | Revision:

root / lib / cmdlib.py @ e1f23243

History | View | Annotate | Download (451.7 kB)

1
#
2
#
3

    
4
# Copyright (C) 2006, 2007, 2008, 2009, 2010, 2011 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,C0302
25

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

    
29
# C0302: since we have waaaay to many lines in this module
30

    
31
import os
32
import os.path
33
import time
34
import re
35
import platform
36
import logging
37
import copy
38
import OpenSSL
39
import socket
40
import tempfile
41
import shutil
42
import itertools
43

    
44
from ganeti import ssh
45
from ganeti import utils
46
from ganeti import errors
47
from ganeti import hypervisor
48
from ganeti import locking
49
from ganeti import constants
50
from ganeti import objects
51
from ganeti import serializer
52
from ganeti import ssconf
53
from ganeti import uidpool
54
from ganeti import compat
55
from ganeti import masterd
56
from ganeti import netutils
57
from ganeti import query
58
from ganeti import qlang
59
from ganeti import opcodes
60
from ganeti import ht
61

    
62
import ganeti.masterd.instance # pylint: disable-msg=W0611
63

    
64

    
65
def _SupportsOob(cfg, node):
66
  """Tells if node supports OOB.
67

68
  @type cfg: L{config.ConfigWriter}
69
  @param cfg: The cluster configuration
70
  @type node: L{objects.Node}
71
  @param node: The node
72
  @return: The OOB script if supported or an empty string otherwise
73

74
  """
75
  return cfg.GetNdParams(node)[constants.ND_OOB_PROGRAM]
76

    
77

    
78
class ResultWithJobs:
79
  """Data container for LU results with jobs.
80

81
  Instances of this class returned from L{LogicalUnit.Exec} will be recognized
82
  by L{mcpu.Processor._ProcessResult}. The latter will then submit the jobs
83
  contained in the C{jobs} attribute and include the job IDs in the opcode
84
  result.
85

86
  """
87
  def __init__(self, jobs, **kwargs):
88
    """Initializes this class.
89

90
    Additional return values can be specified as keyword arguments.
91

92
    @type jobs: list of lists of L{opcode.OpCode}
93
    @param jobs: A list of lists of opcode objects
94

95
    """
96
    self.jobs = jobs
97
    self.other = kwargs
98

    
99

    
100
class LogicalUnit(object):
101
  """Logical Unit base class.
102

103
  Subclasses must follow these rules:
104
    - implement ExpandNames
105
    - implement CheckPrereq (except when tasklets are used)
106
    - implement Exec (except when tasklets are used)
107
    - implement BuildHooksEnv
108
    - implement BuildHooksNodes
109
    - redefine HPATH and HTYPE
110
    - optionally redefine their run requirements:
111
        REQ_BGL: the LU needs to hold the Big Ganeti Lock exclusively
112

113
  Note that all commands require root permissions.
114

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

118
  """
119
  HPATH = None
120
  HTYPE = None
121
  REQ_BGL = True
122

    
123
  def __init__(self, processor, op, context, rpc):
124
    """Constructor for LogicalUnit.
125

126
    This needs to be overridden in derived classes in order to check op
127
    validity.
128

129
    """
130
    self.proc = processor
131
    self.op = op
132
    self.cfg = context.cfg
133
    self.glm = context.glm
134
    self.context = context
135
    self.rpc = rpc
136
    # Dicts used to declare locking needs to mcpu
137
    self.needed_locks = None
138
    self.share_locks = dict.fromkeys(locking.LEVELS, 0)
139
    self.add_locks = {}
140
    self.remove_locks = {}
141
    # Used to force good behavior when calling helper functions
142
    self.recalculate_locks = {}
143
    # logging
144
    self.Log = processor.Log # pylint: disable-msg=C0103
145
    self.LogWarning = processor.LogWarning # pylint: disable-msg=C0103
146
    self.LogInfo = processor.LogInfo # pylint: disable-msg=C0103
147
    self.LogStep = processor.LogStep # pylint: disable-msg=C0103
148
    # support for dry-run
149
    self.dry_run_result = None
150
    # support for generic debug attribute
151
    if (not hasattr(self.op, "debug_level") or
152
        not isinstance(self.op.debug_level, int)):
153
      self.op.debug_level = 0
154

    
155
    # Tasklets
156
    self.tasklets = None
157

    
158
    # Validate opcode parameters and set defaults
159
    self.op.Validate(True)
160

    
161
    self.CheckArguments()
162

    
163
  def CheckArguments(self):
164
    """Check syntactic validity for the opcode arguments.
165

166
    This method is for doing a simple syntactic check and ensure
167
    validity of opcode parameters, without any cluster-related
168
    checks. While the same can be accomplished in ExpandNames and/or
169
    CheckPrereq, doing these separate is better because:
170

171
      - ExpandNames is left as as purely a lock-related function
172
      - CheckPrereq is run after we have acquired locks (and possible
173
        waited for them)
174

175
    The function is allowed to change the self.op attribute so that
176
    later methods can no longer worry about missing parameters.
177

178
    """
179
    pass
180

    
181
  def ExpandNames(self):
182
    """Expand names for this LU.
183

184
    This method is called before starting to execute the opcode, and it should
185
    update all the parameters of the opcode to their canonical form (e.g. a
186
    short node name must be fully expanded after this method has successfully
187
    completed). This way locking, hooks, logging, etc. can work correctly.
188

189
    LUs which implement this method must also populate the self.needed_locks
190
    member, as a dict with lock levels as keys, and a list of needed lock names
191
    as values. Rules:
192

193
      - use an empty dict if you don't need any lock
194
      - if you don't need any lock at a particular level omit that level
195
      - don't put anything for the BGL level
196
      - if you want all locks at a level use locking.ALL_SET as a value
197

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

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

206
    Examples::
207

208
      # Acquire all nodes and one instance
209
      self.needed_locks = {
210
        locking.LEVEL_NODE: locking.ALL_SET,
211
        locking.LEVEL_INSTANCE: ['instance1.example.com'],
212
      }
213
      # Acquire just two nodes
214
      self.needed_locks = {
215
        locking.LEVEL_NODE: ['node1.example.com', 'node2.example.com'],
216
      }
217
      # Acquire no locks
218
      self.needed_locks = {} # No, you can't leave it to the default value None
219

220
    """
221
    # The implementation of this method is mandatory only if the new LU is
222
    # concurrent, so that old LUs don't need to be changed all at the same
223
    # time.
224
    if self.REQ_BGL:
225
      self.needed_locks = {} # Exclusive LUs don't need locks.
226
    else:
227
      raise NotImplementedError
228

    
229
  def DeclareLocks(self, level):
230
    """Declare LU locking needs for a level
231

232
    While most LUs can just declare their locking needs at ExpandNames time,
233
    sometimes there's the need to calculate some locks after having acquired
234
    the ones before. This function is called just before acquiring locks at a
235
    particular level, but after acquiring the ones at lower levels, and permits
236
    such calculations. It can be used to modify self.needed_locks, and by
237
    default it does nothing.
238

239
    This function is only called if you have something already set in
240
    self.needed_locks for the level.
241

242
    @param level: Locking level which is going to be locked
243
    @type level: member of ganeti.locking.LEVELS
244

245
    """
246

    
247
  def CheckPrereq(self):
248
    """Check prerequisites for this LU.
249

250
    This method should check that the prerequisites for the execution
251
    of this LU are fulfilled. It can do internode communication, but
252
    it should be idempotent - no cluster or system changes are
253
    allowed.
254

255
    The method should raise errors.OpPrereqError in case something is
256
    not fulfilled. Its return value is ignored.
257

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

261
    """
262
    if self.tasklets is not None:
263
      for (idx, tl) in enumerate(self.tasklets):
264
        logging.debug("Checking prerequisites for tasklet %s/%s",
265
                      idx + 1, len(self.tasklets))
266
        tl.CheckPrereq()
267
    else:
268
      pass
269

    
270
  def Exec(self, feedback_fn):
271
    """Execute the LU.
272

273
    This method should implement the actual work. It should raise
274
    errors.OpExecError for failures that are somewhat dealt with in
275
    code, or expected.
276

277
    """
278
    if self.tasklets is not None:
279
      for (idx, tl) in enumerate(self.tasklets):
280
        logging.debug("Executing tasklet %s/%s", idx + 1, len(self.tasklets))
281
        tl.Exec(feedback_fn)
282
    else:
283
      raise NotImplementedError
284

    
285
  def BuildHooksEnv(self):
286
    """Build hooks environment for this LU.
287

288
    @rtype: dict
289
    @return: Dictionary containing the environment that will be used for
290
      running the hooks for this LU. The keys of the dict must not be prefixed
291
      with "GANETI_"--that'll be added by the hooks runner. The hooks runner
292
      will extend the environment with additional variables. If no environment
293
      should be defined, an empty dictionary should be returned (not C{None}).
294
    @note: If the C{HPATH} attribute of the LU class is C{None}, this function
295
      will not be called.
296

297
    """
298
    raise NotImplementedError
299

    
300
  def BuildHooksNodes(self):
301
    """Build list of nodes to run LU's hooks.
302

303
    @rtype: tuple; (list, list)
304
    @return: Tuple containing a list of node names on which the hook
305
      should run before the execution and a list of node names on which the
306
      hook should run after the execution. No nodes should be returned as an
307
      empty list (and not None).
308
    @note: If the C{HPATH} attribute of the LU class is C{None}, this function
309
      will not be called.
310

311
    """
312
    raise NotImplementedError
313

    
314
  def HooksCallBack(self, phase, hook_results, feedback_fn, lu_result):
315
    """Notify the LU about the results of its hooks.
316

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

323
    @param phase: one of L{constants.HOOKS_PHASE_POST} or
324
        L{constants.HOOKS_PHASE_PRE}; it denotes the hooks phase
325
    @param hook_results: the results of the multi-node hooks rpc call
326
    @param feedback_fn: function used send feedback back to the caller
327
    @param lu_result: the previous Exec result this LU had, or None
328
        in the PRE phase
329
    @return: the new Exec result, based on the previous result
330
        and hook results
331

332
    """
333
    # API must be kept, thus we ignore the unused argument and could
334
    # be a function warnings
335
    # pylint: disable-msg=W0613,R0201
336
    return lu_result
337

    
338
  def _ExpandAndLockInstance(self):
339
    """Helper function to expand and lock an instance.
340

341
    Many LUs that work on an instance take its name in self.op.instance_name
342
    and need to expand it and then declare the expanded name for locking. This
343
    function does it, and then updates self.op.instance_name to the expanded
344
    name. It also initializes needed_locks as a dict, if this hasn't been done
345
    before.
346

347
    """
348
    if self.needed_locks is None:
349
      self.needed_locks = {}
350
    else:
351
      assert locking.LEVEL_INSTANCE not in self.needed_locks, \
352
        "_ExpandAndLockInstance called with instance-level locks set"
353
    self.op.instance_name = _ExpandInstanceName(self.cfg,
354
                                                self.op.instance_name)
355
    self.needed_locks[locking.LEVEL_INSTANCE] = self.op.instance_name
356

    
357
  def _LockInstancesNodes(self, primary_only=False):
358
    """Helper function to declare instances' nodes for locking.
359

360
    This function should be called after locking one or more instances to lock
361
    their nodes. Its effect is populating self.needed_locks[locking.LEVEL_NODE]
362
    with all primary or secondary nodes for instances already locked and
363
    present in self.needed_locks[locking.LEVEL_INSTANCE].
364

365
    It should be called from DeclareLocks, and for safety only works if
366
    self.recalculate_locks[locking.LEVEL_NODE] is set.
367

368
    In the future it may grow parameters to just lock some instance's nodes, or
369
    to just lock primaries or secondary nodes, if needed.
370

371
    If should be called in DeclareLocks in a way similar to::
372

373
      if level == locking.LEVEL_NODE:
374
        self._LockInstancesNodes()
375

376
    @type primary_only: boolean
377
    @param primary_only: only lock primary nodes of locked instances
378

379
    """
380
    assert locking.LEVEL_NODE in self.recalculate_locks, \
381
      "_LockInstancesNodes helper function called with no nodes to recalculate"
382

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

    
385
    # For now we'll replace self.needed_locks[locking.LEVEL_NODE], but in the
386
    # future we might want to have different behaviors depending on the value
387
    # of self.recalculate_locks[locking.LEVEL_NODE]
388
    wanted_nodes = []
389
    for instance_name in self.glm.list_owned(locking.LEVEL_INSTANCE):
390
      instance = self.context.cfg.GetInstanceInfo(instance_name)
391
      wanted_nodes.append(instance.primary_node)
392
      if not primary_only:
393
        wanted_nodes.extend(instance.secondary_nodes)
394

    
395
    if self.recalculate_locks[locking.LEVEL_NODE] == constants.LOCKS_REPLACE:
396
      self.needed_locks[locking.LEVEL_NODE] = wanted_nodes
397
    elif self.recalculate_locks[locking.LEVEL_NODE] == constants.LOCKS_APPEND:
398
      self.needed_locks[locking.LEVEL_NODE].extend(wanted_nodes)
399

    
400
    del self.recalculate_locks[locking.LEVEL_NODE]
401

    
402

    
403
class NoHooksLU(LogicalUnit): # pylint: disable-msg=W0223
404
  """Simple LU which runs no hooks.
405

406
  This LU is intended as a parent for other LogicalUnits which will
407
  run no hooks, in order to reduce duplicate code.
408

409
  """
410
  HPATH = None
411
  HTYPE = None
412

    
413
  def BuildHooksEnv(self):
414
    """Empty BuildHooksEnv for NoHooksLu.
415

416
    This just raises an error.
417

418
    """
419
    raise AssertionError("BuildHooksEnv called for NoHooksLUs")
420

    
421
  def BuildHooksNodes(self):
422
    """Empty BuildHooksNodes for NoHooksLU.
423

424
    """
425
    raise AssertionError("BuildHooksNodes called for NoHooksLU")
426

    
427

    
428
class Tasklet:
429
  """Tasklet base class.
430

431
  Tasklets are subcomponents for LUs. LUs can consist entirely of tasklets or
432
  they can mix legacy code with tasklets. Locking needs to be done in the LU,
433
  tasklets know nothing about locks.
434

435
  Subclasses must follow these rules:
436
    - Implement CheckPrereq
437
    - Implement Exec
438

439
  """
440
  def __init__(self, lu):
441
    self.lu = lu
442

    
443
    # Shortcuts
444
    self.cfg = lu.cfg
445
    self.rpc = lu.rpc
446

    
447
  def CheckPrereq(self):
448
    """Check prerequisites for this tasklets.
449

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

454
    The method should raise errors.OpPrereqError in case something is not
455
    fulfilled. Its return value is ignored.
456

457
    This method should also update all parameters to their canonical form if it
458
    hasn't been done before.
459

460
    """
461
    pass
462

    
463
  def Exec(self, feedback_fn):
464
    """Execute the tasklet.
465

466
    This method should implement the actual work. It should raise
467
    errors.OpExecError for failures that are somewhat dealt with in code, or
468
    expected.
469

470
    """
471
    raise NotImplementedError
472

    
473

    
474
class _QueryBase:
475
  """Base for query utility classes.
476

477
  """
478
  #: Attribute holding field definitions
479
  FIELDS = None
480

    
481
  def __init__(self, filter_, fields, use_locking):
482
    """Initializes this class.
483

484
    """
485
    self.use_locking = use_locking
486

    
487
    self.query = query.Query(self.FIELDS, fields, filter_=filter_,
488
                             namefield="name")
489
    self.requested_data = self.query.RequestedData()
490
    self.names = self.query.RequestedNames()
491

    
492
    # Sort only if no names were requested
493
    self.sort_by_name = not self.names
494

    
495
    self.do_locking = None
496
    self.wanted = None
497

    
498
  def _GetNames(self, lu, all_names, lock_level):
499
    """Helper function to determine names asked for in the query.
500

501
    """
502
    if self.do_locking:
503
      names = lu.glm.list_owned(lock_level)
504
    else:
505
      names = all_names
506

    
507
    if self.wanted == locking.ALL_SET:
508
      assert not self.names
509
      # caller didn't specify names, so ordering is not important
510
      return utils.NiceSort(names)
511

    
512
    # caller specified names and we must keep the same order
513
    assert self.names
514
    assert not self.do_locking or lu.glm.is_owned(lock_level)
515

    
516
    missing = set(self.wanted).difference(names)
517
    if missing:
518
      raise errors.OpExecError("Some items were removed before retrieving"
519
                               " their data: %s" % missing)
520

    
521
    # Return expanded names
522
    return self.wanted
523

    
524
  def ExpandNames(self, lu):
525
    """Expand names for this query.
526

527
    See L{LogicalUnit.ExpandNames}.
528

529
    """
530
    raise NotImplementedError()
531

    
532
  def DeclareLocks(self, lu, level):
533
    """Declare locks for this query.
534

535
    See L{LogicalUnit.DeclareLocks}.
536

537
    """
538
    raise NotImplementedError()
539

    
540
  def _GetQueryData(self, lu):
541
    """Collects all data for this query.
542

543
    @return: Query data object
544

545
    """
546
    raise NotImplementedError()
547

    
548
  def NewStyleQuery(self, lu):
549
    """Collect data and execute query.
550

551
    """
552
    return query.GetQueryResponse(self.query, self._GetQueryData(lu),
553
                                  sort_by_name=self.sort_by_name)
554

    
555
  def OldStyleQuery(self, lu):
556
    """Collect data and execute query.
557

558
    """
559
    return self.query.OldStyleQuery(self._GetQueryData(lu),
560
                                    sort_by_name=self.sort_by_name)
561

    
562

    
563
def _GetWantedNodes(lu, nodes):
564
  """Returns list of checked and expanded node names.
565

566
  @type lu: L{LogicalUnit}
567
  @param lu: the logical unit on whose behalf we execute
568
  @type nodes: list
569
  @param nodes: list of node names or None for all nodes
570
  @rtype: list
571
  @return: the list of nodes, sorted
572
  @raise errors.ProgrammerError: if the nodes parameter is wrong type
573

574
  """
575
  if nodes:
576
    return [_ExpandNodeName(lu.cfg, name) for name in nodes]
577

    
578
  return utils.NiceSort(lu.cfg.GetNodeList())
579

    
580

    
581
def _GetWantedInstances(lu, instances):
582
  """Returns list of checked and expanded instance names.
583

584
  @type lu: L{LogicalUnit}
585
  @param lu: the logical unit on whose behalf we execute
586
  @type instances: list
587
  @param instances: list of instance names or None for all instances
588
  @rtype: list
589
  @return: the list of instances, sorted
590
  @raise errors.OpPrereqError: if the instances parameter is wrong type
591
  @raise errors.OpPrereqError: if any of the passed instances is not found
592

593
  """
594
  if instances:
595
    wanted = [_ExpandInstanceName(lu.cfg, name) for name in instances]
596
  else:
597
    wanted = utils.NiceSort(lu.cfg.GetInstanceList())
598
  return wanted
599

    
600

    
601
def _GetUpdatedParams(old_params, update_dict,
602
                      use_default=True, use_none=False):
603
  """Return the new version of a parameter dictionary.
604

605
  @type old_params: dict
606
  @param old_params: old parameters
607
  @type update_dict: dict
608
  @param update_dict: dict containing new parameter values, or
609
      constants.VALUE_DEFAULT to reset the parameter to its default
610
      value
611
  @param use_default: boolean
612
  @type use_default: whether to recognise L{constants.VALUE_DEFAULT}
613
      values as 'to be deleted' values
614
  @param use_none: boolean
615
  @type use_none: whether to recognise C{None} values as 'to be
616
      deleted' values
617
  @rtype: dict
618
  @return: the new parameter dictionary
619

620
  """
621
  params_copy = copy.deepcopy(old_params)
622
  for key, val in update_dict.iteritems():
623
    if ((use_default and val == constants.VALUE_DEFAULT) or
624
        (use_none and val is None)):
625
      try:
626
        del params_copy[key]
627
      except KeyError:
628
        pass
629
    else:
630
      params_copy[key] = val
631
  return params_copy
632

    
633

    
634
def _ReleaseLocks(lu, level, names=None, keep=None):
635
  """Releases locks owned by an LU.
636

637
  @type lu: L{LogicalUnit}
638
  @param level: Lock level
639
  @type names: list or None
640
  @param names: Names of locks to release
641
  @type keep: list or None
642
  @param keep: Names of locks to retain
643

644
  """
645
  assert not (keep is not None and names is not None), \
646
         "Only one of the 'names' and the 'keep' parameters can be given"
647

    
648
  if names is not None:
649
    should_release = names.__contains__
650
  elif keep:
651
    should_release = lambda name: name not in keep
652
  else:
653
    should_release = None
654

    
655
  if should_release:
656
    retain = []
657
    release = []
658

    
659
    # Determine which locks to release
660
    for name in lu.glm.list_owned(level):
661
      if should_release(name):
662
        release.append(name)
663
      else:
664
        retain.append(name)
665

    
666
    assert len(lu.glm.list_owned(level)) == (len(retain) + len(release))
667

    
668
    # Release just some locks
669
    lu.glm.release(level, names=release)
670

    
671
    assert frozenset(lu.glm.list_owned(level)) == frozenset(retain)
672
  else:
673
    # Release everything
674
    lu.glm.release(level)
675

    
676
    assert not lu.glm.is_owned(level), "No locks should be owned"
677

    
678

    
679
def _RunPostHook(lu, node_name):
680
  """Runs the post-hook for an opcode on a single node.
681

682
  """
683
  hm = lu.proc.hmclass(lu.rpc.call_hooks_runner, lu)
684
  try:
685
    hm.RunPhase(constants.HOOKS_PHASE_POST, nodes=[node_name])
686
  except:
687
    # pylint: disable-msg=W0702
688
    lu.LogWarning("Errors occurred running hooks on %s" % node_name)
689

    
690

    
691
def _CheckOutputFields(static, dynamic, selected):
692
  """Checks whether all selected fields are valid.
693

694
  @type static: L{utils.FieldSet}
695
  @param static: static fields set
696
  @type dynamic: L{utils.FieldSet}
697
  @param dynamic: dynamic fields set
698

699
  """
700
  f = utils.FieldSet()
701
  f.Extend(static)
702
  f.Extend(dynamic)
703

    
704
  delta = f.NonMatching(selected)
705
  if delta:
706
    raise errors.OpPrereqError("Unknown output fields selected: %s"
707
                               % ",".join(delta), errors.ECODE_INVAL)
708

    
709

    
710
def _CheckGlobalHvParams(params):
711
  """Validates that given hypervisor params are not global ones.
712

713
  This will ensure that instances don't get customised versions of
714
  global params.
715

716
  """
717
  used_globals = constants.HVC_GLOBALS.intersection(params)
718
  if used_globals:
719
    msg = ("The following hypervisor parameters are global and cannot"
720
           " be customized at instance level, please modify them at"
721
           " cluster level: %s" % utils.CommaJoin(used_globals))
722
    raise errors.OpPrereqError(msg, errors.ECODE_INVAL)
723

    
724

    
725
def _CheckNodeOnline(lu, node, msg=None):
726
  """Ensure that a given node is online.
727

728
  @param lu: the LU on behalf of which we make the check
729
  @param node: the node to check
730
  @param msg: if passed, should be a message to replace the default one
731
  @raise errors.OpPrereqError: if the node is offline
732

733
  """
734
  if msg is None:
735
    msg = "Can't use offline node"
736
  if lu.cfg.GetNodeInfo(node).offline:
737
    raise errors.OpPrereqError("%s: %s" % (msg, node), errors.ECODE_STATE)
738

    
739

    
740
def _CheckNodeNotDrained(lu, node):
741
  """Ensure that a given node is not drained.
742

743
  @param lu: the LU on behalf of which we make the check
744
  @param node: the node to check
745
  @raise errors.OpPrereqError: if the node is drained
746

747
  """
748
  if lu.cfg.GetNodeInfo(node).drained:
749
    raise errors.OpPrereqError("Can't use drained node %s" % node,
750
                               errors.ECODE_STATE)
751

    
752

    
753
def _CheckNodeVmCapable(lu, node):
754
  """Ensure that a given node is vm capable.
755

756
  @param lu: the LU on behalf of which we make the check
757
  @param node: the node to check
758
  @raise errors.OpPrereqError: if the node is not vm capable
759

760
  """
761
  if not lu.cfg.GetNodeInfo(node).vm_capable:
762
    raise errors.OpPrereqError("Can't use non-vm_capable node %s" % node,
763
                               errors.ECODE_STATE)
764

    
765

    
766
def _CheckNodeHasOS(lu, node, os_name, force_variant):
767
  """Ensure that a node supports a given OS.
768

769
  @param lu: the LU on behalf of which we make the check
770
  @param node: the node to check
771
  @param os_name: the OS to query about
772
  @param force_variant: whether to ignore variant errors
773
  @raise errors.OpPrereqError: if the node is not supporting the OS
774

775
  """
776
  result = lu.rpc.call_os_get(node, os_name)
777
  result.Raise("OS '%s' not in supported OS list for node %s" %
778
               (os_name, node),
779
               prereq=True, ecode=errors.ECODE_INVAL)
780
  if not force_variant:
781
    _CheckOSVariant(result.payload, os_name)
782

    
783

    
784
def _CheckNodeHasSecondaryIP(lu, node, secondary_ip, prereq):
785
  """Ensure that a node has the given secondary ip.
786

787
  @type lu: L{LogicalUnit}
788
  @param lu: the LU on behalf of which we make the check
789
  @type node: string
790
  @param node: the node to check
791
  @type secondary_ip: string
792
  @param secondary_ip: the ip to check
793
  @type prereq: boolean
794
  @param prereq: whether to throw a prerequisite or an execute error
795
  @raise errors.OpPrereqError: if the node doesn't have the ip, and prereq=True
796
  @raise errors.OpExecError: if the node doesn't have the ip, and prereq=False
797

798
  """
799
  result = lu.rpc.call_node_has_ip_address(node, secondary_ip)
800
  result.Raise("Failure checking secondary ip on node %s" % node,
801
               prereq=prereq, ecode=errors.ECODE_ENVIRON)
802
  if not result.payload:
803
    msg = ("Node claims it doesn't have the secondary ip you gave (%s),"
804
           " please fix and re-run this command" % secondary_ip)
805
    if prereq:
806
      raise errors.OpPrereqError(msg, errors.ECODE_ENVIRON)
807
    else:
808
      raise errors.OpExecError(msg)
809

    
810

    
811
def _GetClusterDomainSecret():
812
  """Reads the cluster domain secret.
813

814
  """
815
  return utils.ReadOneLineFile(constants.CLUSTER_DOMAIN_SECRET_FILE,
816
                               strict=True)
817

    
818

    
819
def _CheckInstanceDown(lu, instance, reason):
820
  """Ensure that an instance is not running."""
821
  if instance.admin_up:
822
    raise errors.OpPrereqError("Instance %s is marked to be up, %s" %
823
                               (instance.name, reason), errors.ECODE_STATE)
824

    
825
  pnode = instance.primary_node
826
  ins_l = lu.rpc.call_instance_list([pnode], [instance.hypervisor])[pnode]
827
  ins_l.Raise("Can't contact node %s for instance information" % pnode,
828
              prereq=True, ecode=errors.ECODE_ENVIRON)
829

    
830
  if instance.name in ins_l.payload:
831
    raise errors.OpPrereqError("Instance %s is running, %s" %
832
                               (instance.name, reason), errors.ECODE_STATE)
833

    
834

    
835
def _ExpandItemName(fn, name, kind):
836
  """Expand an item name.
837

838
  @param fn: the function to use for expansion
839
  @param name: requested item name
840
  @param kind: text description ('Node' or 'Instance')
841
  @return: the resolved (full) name
842
  @raise errors.OpPrereqError: if the item is not found
843

844
  """
845
  full_name = fn(name)
846
  if full_name is None:
847
    raise errors.OpPrereqError("%s '%s' not known" % (kind, name),
848
                               errors.ECODE_NOENT)
849
  return full_name
850

    
851

    
852
def _ExpandNodeName(cfg, name):
853
  """Wrapper over L{_ExpandItemName} for nodes."""
854
  return _ExpandItemName(cfg.ExpandNodeName, name, "Node")
855

    
856

    
857
def _ExpandInstanceName(cfg, name):
858
  """Wrapper over L{_ExpandItemName} for instance."""
859
  return _ExpandItemName(cfg.ExpandInstanceName, name, "Instance")
860

    
861

    
862
def _BuildInstanceHookEnv(name, primary_node, secondary_nodes, os_type, status,
863
                          memory, vcpus, nics, disk_template, disks,
864
                          bep, hvp, hypervisor_name, tags):
865
  """Builds instance related env variables for hooks
866

867
  This builds the hook environment from individual variables.
868

869
  @type name: string
870
  @param name: the name of the instance
871
  @type primary_node: string
872
  @param primary_node: the name of the instance's primary node
873
  @type secondary_nodes: list
874
  @param secondary_nodes: list of secondary nodes as strings
875
  @type os_type: string
876
  @param os_type: the name of the instance's OS
877
  @type status: boolean
878
  @param status: the should_run status of the instance
879
  @type memory: string
880
  @param memory: the memory size of the instance
881
  @type vcpus: string
882
  @param vcpus: the count of VCPUs the instance has
883
  @type nics: list
884
  @param nics: list of tuples (ip, mac, mode, link) representing
885
      the NICs the instance has
886
  @type disk_template: string
887
  @param disk_template: the disk template of the instance
888
  @type disks: list
889
  @param disks: the list of (size, mode) pairs
890
  @type bep: dict
891
  @param bep: the backend parameters for the instance
892
  @type hvp: dict
893
  @param hvp: the hypervisor parameters for the instance
894
  @type hypervisor_name: string
895
  @param hypervisor_name: the hypervisor for the instance
896
  @type tags: list
897
  @param tags: list of instance tags as strings
898
  @rtype: dict
899
  @return: the hook environment for this instance
900

901
  """
902
  if status:
903
    str_status = "up"
904
  else:
905
    str_status = "down"
906
  env = {
907
    "OP_TARGET": name,
908
    "INSTANCE_NAME": name,
909
    "INSTANCE_PRIMARY": primary_node,
910
    "INSTANCE_SECONDARIES": " ".join(secondary_nodes),
911
    "INSTANCE_OS_TYPE": os_type,
912
    "INSTANCE_STATUS": str_status,
913
    "INSTANCE_MEMORY": memory,
914
    "INSTANCE_VCPUS": vcpus,
915
    "INSTANCE_DISK_TEMPLATE": disk_template,
916
    "INSTANCE_HYPERVISOR": hypervisor_name,
917
  }
918

    
919
  if nics:
920
    nic_count = len(nics)
921
    for idx, (ip, mac, mode, link) in enumerate(nics):
922
      if ip is None:
923
        ip = ""
924
      env["INSTANCE_NIC%d_IP" % idx] = ip
925
      env["INSTANCE_NIC%d_MAC" % idx] = mac
926
      env["INSTANCE_NIC%d_MODE" % idx] = mode
927
      env["INSTANCE_NIC%d_LINK" % idx] = link
928
      if mode == constants.NIC_MODE_BRIDGED:
929
        env["INSTANCE_NIC%d_BRIDGE" % idx] = link
930
  else:
931
    nic_count = 0
932

    
933
  env["INSTANCE_NIC_COUNT"] = nic_count
934

    
935
  if disks:
936
    disk_count = len(disks)
937
    for idx, (size, mode) in enumerate(disks):
938
      env["INSTANCE_DISK%d_SIZE" % idx] = size
939
      env["INSTANCE_DISK%d_MODE" % idx] = mode
940
  else:
941
    disk_count = 0
942

    
943
  env["INSTANCE_DISK_COUNT"] = disk_count
944

    
945
  if not tags:
946
    tags = []
947

    
948
  env["INSTANCE_TAGS"] = " ".join(tags)
949

    
950
  for source, kind in [(bep, "BE"), (hvp, "HV")]:
951
    for key, value in source.items():
952
      env["INSTANCE_%s_%s" % (kind, key)] = value
953

    
954
  return env
955

    
956

    
957
def _NICListToTuple(lu, nics):
958
  """Build a list of nic information tuples.
959

960
  This list is suitable to be passed to _BuildInstanceHookEnv or as a return
961
  value in LUInstanceQueryData.
962

963
  @type lu:  L{LogicalUnit}
964
  @param lu: the logical unit on whose behalf we execute
965
  @type nics: list of L{objects.NIC}
966
  @param nics: list of nics to convert to hooks tuples
967

968
  """
969
  hooks_nics = []
970
  cluster = lu.cfg.GetClusterInfo()
971
  for nic in nics:
972
    ip = nic.ip
973
    mac = nic.mac
974
    filled_params = cluster.SimpleFillNIC(nic.nicparams)
975
    mode = filled_params[constants.NIC_MODE]
976
    link = filled_params[constants.NIC_LINK]
977
    hooks_nics.append((ip, mac, mode, link))
978
  return hooks_nics
979

    
980

    
981
def _BuildInstanceHookEnvByObject(lu, instance, override=None):
982
  """Builds instance related env variables for hooks from an object.
983

984
  @type lu: L{LogicalUnit}
985
  @param lu: the logical unit on whose behalf we execute
986
  @type instance: L{objects.Instance}
987
  @param instance: the instance for which we should build the
988
      environment
989
  @type override: dict
990
  @param override: dictionary with key/values that will override
991
      our values
992
  @rtype: dict
993
  @return: the hook environment dictionary
994

995
  """
996
  cluster = lu.cfg.GetClusterInfo()
997
  bep = cluster.FillBE(instance)
998
  hvp = cluster.FillHV(instance)
999
  args = {
1000
    'name': instance.name,
1001
    'primary_node': instance.primary_node,
1002
    'secondary_nodes': instance.secondary_nodes,
1003
    'os_type': instance.os,
1004
    'status': instance.admin_up,
1005
    'memory': bep[constants.BE_MEMORY],
1006
    'vcpus': bep[constants.BE_VCPUS],
1007
    'nics': _NICListToTuple(lu, instance.nics),
1008
    'disk_template': instance.disk_template,
1009
    'disks': [(disk.size, disk.mode) for disk in instance.disks],
1010
    'bep': bep,
1011
    'hvp': hvp,
1012
    'hypervisor_name': instance.hypervisor,
1013
    'tags': instance.tags,
1014
  }
1015
  if override:
1016
    args.update(override)
1017
  return _BuildInstanceHookEnv(**args) # pylint: disable-msg=W0142
1018

    
1019

    
1020
def _AdjustCandidatePool(lu, exceptions):
1021
  """Adjust the candidate pool after node operations.
1022

1023
  """
1024
  mod_list = lu.cfg.MaintainCandidatePool(exceptions)
1025
  if mod_list:
1026
    lu.LogInfo("Promoted nodes to master candidate role: %s",
1027
               utils.CommaJoin(node.name for node in mod_list))
1028
    for name in mod_list:
1029
      lu.context.ReaddNode(name)
1030
  mc_now, mc_max, _ = lu.cfg.GetMasterCandidateStats(exceptions)
1031
  if mc_now > mc_max:
1032
    lu.LogInfo("Note: more nodes are candidates (%d) than desired (%d)" %
1033
               (mc_now, mc_max))
1034

    
1035

    
1036
def _DecideSelfPromotion(lu, exceptions=None):
1037
  """Decide whether I should promote myself as a master candidate.
1038

1039
  """
1040
  cp_size = lu.cfg.GetClusterInfo().candidate_pool_size
1041
  mc_now, mc_should, _ = lu.cfg.GetMasterCandidateStats(exceptions)
1042
  # the new node will increase mc_max with one, so:
1043
  mc_should = min(mc_should + 1, cp_size)
1044
  return mc_now < mc_should
1045

    
1046

    
1047
def _CheckNicsBridgesExist(lu, target_nics, target_node):
1048
  """Check that the brigdes needed by a list of nics exist.
1049

1050
  """
1051
  cluster = lu.cfg.GetClusterInfo()
1052
  paramslist = [cluster.SimpleFillNIC(nic.nicparams) for nic in target_nics]
1053
  brlist = [params[constants.NIC_LINK] for params in paramslist
1054
            if params[constants.NIC_MODE] == constants.NIC_MODE_BRIDGED]
1055
  if brlist:
1056
    result = lu.rpc.call_bridges_exist(target_node, brlist)
1057
    result.Raise("Error checking bridges on destination node '%s'" %
1058
                 target_node, prereq=True, ecode=errors.ECODE_ENVIRON)
1059

    
1060

    
1061
def _CheckInstanceBridgesExist(lu, instance, node=None):
1062
  """Check that the brigdes needed by an instance exist.
1063

1064
  """
1065
  if node is None:
1066
    node = instance.primary_node
1067
  _CheckNicsBridgesExist(lu, instance.nics, node)
1068

    
1069

    
1070
def _CheckOSVariant(os_obj, name):
1071
  """Check whether an OS name conforms to the os variants specification.
1072

1073
  @type os_obj: L{objects.OS}
1074
  @param os_obj: OS object to check
1075
  @type name: string
1076
  @param name: OS name passed by the user, to check for validity
1077

1078
  """
1079
  if not os_obj.supported_variants:
1080
    return
1081
  variant = objects.OS.GetVariant(name)
1082
  if not variant:
1083
    raise errors.OpPrereqError("OS name must include a variant",
1084
                               errors.ECODE_INVAL)
1085

    
1086
  if variant not in os_obj.supported_variants:
1087
    raise errors.OpPrereqError("Unsupported OS variant", errors.ECODE_INVAL)
1088

    
1089

    
1090
def _GetNodeInstancesInner(cfg, fn):
1091
  return [i for i in cfg.GetAllInstancesInfo().values() if fn(i)]
1092

    
1093

    
1094
def _GetNodeInstances(cfg, node_name):
1095
  """Returns a list of all primary and secondary instances on a node.
1096

1097
  """
1098

    
1099
  return _GetNodeInstancesInner(cfg, lambda inst: node_name in inst.all_nodes)
1100

    
1101

    
1102
def _GetNodePrimaryInstances(cfg, node_name):
1103
  """Returns primary instances on a node.
1104

1105
  """
1106
  return _GetNodeInstancesInner(cfg,
1107
                                lambda inst: node_name == inst.primary_node)
1108

    
1109

    
1110
def _GetNodeSecondaryInstances(cfg, node_name):
1111
  """Returns secondary instances on a node.
1112

1113
  """
1114
  return _GetNodeInstancesInner(cfg,
1115
                                lambda inst: node_name in inst.secondary_nodes)
1116

    
1117

    
1118
def _GetStorageTypeArgs(cfg, storage_type):
1119
  """Returns the arguments for a storage type.
1120

1121
  """
1122
  # Special case for file storage
1123
  if storage_type == constants.ST_FILE:
1124
    # storage.FileStorage wants a list of storage directories
1125
    return [[cfg.GetFileStorageDir(), cfg.GetSharedFileStorageDir()]]
1126

    
1127
  return []
1128

    
1129

    
1130
def _FindFaultyInstanceDisks(cfg, rpc, instance, node_name, prereq):
1131
  faulty = []
1132

    
1133
  for dev in instance.disks:
1134
    cfg.SetDiskID(dev, node_name)
1135

    
1136
  result = rpc.call_blockdev_getmirrorstatus(node_name, instance.disks)
1137
  result.Raise("Failed to get disk status from node %s" % node_name,
1138
               prereq=prereq, ecode=errors.ECODE_ENVIRON)
1139

    
1140
  for idx, bdev_status in enumerate(result.payload):
1141
    if bdev_status and bdev_status.ldisk_status == constants.LDS_FAULTY:
1142
      faulty.append(idx)
1143

    
1144
  return faulty
1145

    
1146

    
1147
def _CheckIAllocatorOrNode(lu, iallocator_slot, node_slot):
1148
  """Check the sanity of iallocator and node arguments and use the
1149
  cluster-wide iallocator if appropriate.
1150

1151
  Check that at most one of (iallocator, node) is specified. If none is
1152
  specified, then the LU's opcode's iallocator slot is filled with the
1153
  cluster-wide default iallocator.
1154

1155
  @type iallocator_slot: string
1156
  @param iallocator_slot: the name of the opcode iallocator slot
1157
  @type node_slot: string
1158
  @param node_slot: the name of the opcode target node slot
1159

1160
  """
1161
  node = getattr(lu.op, node_slot, None)
1162
  iallocator = getattr(lu.op, iallocator_slot, None)
1163

    
1164
  if node is not None and iallocator is not None:
1165
    raise errors.OpPrereqError("Do not specify both, iallocator and node",
1166
                               errors.ECODE_INVAL)
1167
  elif node is None and iallocator is None:
1168
    default_iallocator = lu.cfg.GetDefaultIAllocator()
1169
    if default_iallocator:
1170
      setattr(lu.op, iallocator_slot, default_iallocator)
1171
    else:
1172
      raise errors.OpPrereqError("No iallocator or node given and no"
1173
                                 " cluster-wide default iallocator found;"
1174
                                 " please specify either an iallocator or a"
1175
                                 " node, or set a cluster-wide default"
1176
                                 " iallocator")
1177

    
1178

    
1179
class LUClusterPostInit(LogicalUnit):
1180
  """Logical unit for running hooks after cluster initialization.
1181

1182
  """
1183
  HPATH = "cluster-init"
1184
  HTYPE = constants.HTYPE_CLUSTER
1185

    
1186
  def BuildHooksEnv(self):
1187
    """Build hooks env.
1188

1189
    """
1190
    return {
1191
      "OP_TARGET": self.cfg.GetClusterName(),
1192
      }
1193

    
1194
  def BuildHooksNodes(self):
1195
    """Build hooks nodes.
1196

1197
    """
1198
    return ([], [self.cfg.GetMasterNode()])
1199

    
1200
  def Exec(self, feedback_fn):
1201
    """Nothing to do.
1202

1203
    """
1204
    return True
1205

    
1206

    
1207
class LUClusterDestroy(LogicalUnit):
1208
  """Logical unit for destroying the cluster.
1209

1210
  """
1211
  HPATH = "cluster-destroy"
1212
  HTYPE = constants.HTYPE_CLUSTER
1213

    
1214
  def BuildHooksEnv(self):
1215
    """Build hooks env.
1216

1217
    """
1218
    return {
1219
      "OP_TARGET": self.cfg.GetClusterName(),
1220
      }
1221

    
1222
  def BuildHooksNodes(self):
1223
    """Build hooks nodes.
1224

1225
    """
1226
    return ([], [])
1227

    
1228
  def CheckPrereq(self):
1229
    """Check prerequisites.
1230

1231
    This checks whether the cluster is empty.
1232

1233
    Any errors are signaled by raising errors.OpPrereqError.
1234

1235
    """
1236
    master = self.cfg.GetMasterNode()
1237

    
1238
    nodelist = self.cfg.GetNodeList()
1239
    if len(nodelist) != 1 or nodelist[0] != master:
1240
      raise errors.OpPrereqError("There are still %d node(s) in"
1241
                                 " this cluster." % (len(nodelist) - 1),
1242
                                 errors.ECODE_INVAL)
1243
    instancelist = self.cfg.GetInstanceList()
1244
    if instancelist:
1245
      raise errors.OpPrereqError("There are still %d instance(s) in"
1246
                                 " this cluster." % len(instancelist),
1247
                                 errors.ECODE_INVAL)
1248

    
1249
  def Exec(self, feedback_fn):
1250
    """Destroys the cluster.
1251

1252
    """
1253
    master = self.cfg.GetMasterNode()
1254

    
1255
    # Run post hooks on master node before it's removed
1256
    _RunPostHook(self, master)
1257

    
1258
    result = self.rpc.call_node_stop_master(master, False)
1259
    result.Raise("Could not disable the master role")
1260

    
1261
    return master
1262

    
1263

    
1264
def _VerifyCertificate(filename):
1265
  """Verifies a certificate for L{LUClusterVerifyConfig}.
1266

1267
  @type filename: string
1268
  @param filename: Path to PEM file
1269

1270
  """
1271
  try:
1272
    cert = OpenSSL.crypto.load_certificate(OpenSSL.crypto.FILETYPE_PEM,
1273
                                           utils.ReadFile(filename))
1274
  except Exception, err: # pylint: disable-msg=W0703
1275
    return (LUClusterVerifyConfig.ETYPE_ERROR,
1276
            "Failed to load X509 certificate %s: %s" % (filename, err))
1277

    
1278
  (errcode, msg) = \
1279
    utils.VerifyX509Certificate(cert, constants.SSL_CERT_EXPIRATION_WARN,
1280
                                constants.SSL_CERT_EXPIRATION_ERROR)
1281

    
1282
  if msg:
1283
    fnamemsg = "While verifying %s: %s" % (filename, msg)
1284
  else:
1285
    fnamemsg = None
1286

    
1287
  if errcode is None:
1288
    return (None, fnamemsg)
1289
  elif errcode == utils.CERT_WARNING:
1290
    return (LUClusterVerifyConfig.ETYPE_WARNING, fnamemsg)
1291
  elif errcode == utils.CERT_ERROR:
1292
    return (LUClusterVerifyConfig.ETYPE_ERROR, fnamemsg)
1293

    
1294
  raise errors.ProgrammerError("Unhandled certificate error code %r" % errcode)
1295

    
1296

    
1297
def _GetAllHypervisorParameters(cluster, instances):
1298
  """Compute the set of all hypervisor parameters.
1299

1300
  @type cluster: L{objects.Cluster}
1301
  @param cluster: the cluster object
1302
  @param instances: list of L{objects.Instance}
1303
  @param instances: additional instances from which to obtain parameters
1304
  @rtype: list of (origin, hypervisor, parameters)
1305
  @return: a list with all parameters found, indicating the hypervisor they
1306
       apply to, and the origin (can be "cluster", "os X", or "instance Y")
1307

1308
  """
1309
  hvp_data = []
1310

    
1311
  for hv_name in cluster.enabled_hypervisors:
1312
    hvp_data.append(("cluster", hv_name, cluster.GetHVDefaults(hv_name)))
1313

    
1314
  for os_name, os_hvp in cluster.os_hvp.items():
1315
    for hv_name, hv_params in os_hvp.items():
1316
      if hv_params:
1317
        full_params = cluster.GetHVDefaults(hv_name, os_name=os_name)
1318
        hvp_data.append(("os %s" % os_name, hv_name, full_params))
1319

    
1320
  # TODO: collapse identical parameter values in a single one
1321
  for instance in instances:
1322
    if instance.hvparams:
1323
      hvp_data.append(("instance %s" % instance.name, instance.hypervisor,
1324
                       cluster.FillHV(instance)))
1325

    
1326
  return hvp_data
1327

    
1328

    
1329
class _VerifyErrors(object):
1330
  """Mix-in for cluster/group verify LUs.
1331

1332
  It provides _Error and _ErrorIf, and updates the self.bad boolean. (Expects
1333
  self.op and self._feedback_fn to be available.)
1334

1335
  """
1336
  TCLUSTER = "cluster"
1337
  TNODE = "node"
1338
  TINSTANCE = "instance"
1339

    
1340
  ECLUSTERCFG = (TCLUSTER, "ECLUSTERCFG")
1341
  ECLUSTERCERT = (TCLUSTER, "ECLUSTERCERT")
1342
  ECLUSTERFILECHECK = (TCLUSTER, "ECLUSTERFILECHECK")
1343
  ECLUSTERDANGLINGNODES = (TNODE, "ECLUSTERDANGLINGNODES")
1344
  ECLUSTERDANGLINGINST = (TNODE, "ECLUSTERDANGLINGINST")
1345
  EINSTANCEBADNODE = (TINSTANCE, "EINSTANCEBADNODE")
1346
  EINSTANCEDOWN = (TINSTANCE, "EINSTANCEDOWN")
1347
  EINSTANCELAYOUT = (TINSTANCE, "EINSTANCELAYOUT")
1348
  EINSTANCEMISSINGDISK = (TINSTANCE, "EINSTANCEMISSINGDISK")
1349
  EINSTANCEFAULTYDISK = (TINSTANCE, "EINSTANCEFAULTYDISK")
1350
  EINSTANCEWRONGNODE = (TINSTANCE, "EINSTANCEWRONGNODE")
1351
  EINSTANCESPLITGROUPS = (TINSTANCE, "EINSTANCESPLITGROUPS")
1352
  ENODEDRBD = (TNODE, "ENODEDRBD")
1353
  ENODEDRBDHELPER = (TNODE, "ENODEDRBDHELPER")
1354
  ENODEFILECHECK = (TNODE, "ENODEFILECHECK")
1355
  ENODEHOOKS = (TNODE, "ENODEHOOKS")
1356
  ENODEHV = (TNODE, "ENODEHV")
1357
  ENODELVM = (TNODE, "ENODELVM")
1358
  ENODEN1 = (TNODE, "ENODEN1")
1359
  ENODENET = (TNODE, "ENODENET")
1360
  ENODEOS = (TNODE, "ENODEOS")
1361
  ENODEORPHANINSTANCE = (TNODE, "ENODEORPHANINSTANCE")
1362
  ENODEORPHANLV = (TNODE, "ENODEORPHANLV")
1363
  ENODERPC = (TNODE, "ENODERPC")
1364
  ENODESSH = (TNODE, "ENODESSH")
1365
  ENODEVERSION = (TNODE, "ENODEVERSION")
1366
  ENODESETUP = (TNODE, "ENODESETUP")
1367
  ENODETIME = (TNODE, "ENODETIME")
1368
  ENODEOOBPATH = (TNODE, "ENODEOOBPATH")
1369

    
1370
  ETYPE_FIELD = "code"
1371
  ETYPE_ERROR = "ERROR"
1372
  ETYPE_WARNING = "WARNING"
1373

    
1374
  def _Error(self, ecode, item, msg, *args, **kwargs):
1375
    """Format an error message.
1376

1377
    Based on the opcode's error_codes parameter, either format a
1378
    parseable error code, or a simpler error string.
1379

1380
    This must be called only from Exec and functions called from Exec.
1381

1382
    """
1383
    ltype = kwargs.get(self.ETYPE_FIELD, self.ETYPE_ERROR)
1384
    itype, etxt = ecode
1385
    # first complete the msg
1386
    if args:
1387
      msg = msg % args
1388
    # then format the whole message
1389
    if self.op.error_codes: # This is a mix-in. pylint: disable-msg=E1101
1390
      msg = "%s:%s:%s:%s:%s" % (ltype, etxt, itype, item, msg)
1391
    else:
1392
      if item:
1393
        item = " " + item
1394
      else:
1395
        item = ""
1396
      msg = "%s: %s%s: %s" % (ltype, itype, item, msg)
1397
    # and finally report it via the feedback_fn
1398
    self._feedback_fn("  - %s" % msg) # Mix-in. pylint: disable-msg=E1101
1399

    
1400
  def _ErrorIf(self, cond, *args, **kwargs):
1401
    """Log an error message if the passed condition is True.
1402

1403
    """
1404
    cond = (bool(cond)
1405
            or self.op.debug_simulate_errors) # pylint: disable-msg=E1101
1406
    if cond:
1407
      self._Error(*args, **kwargs)
1408
    # do not mark the operation as failed for WARN cases only
1409
    if kwargs.get(self.ETYPE_FIELD, self.ETYPE_ERROR) == self.ETYPE_ERROR:
1410
      self.bad = self.bad or cond
1411

    
1412

    
1413
class LUClusterVerifyConfig(NoHooksLU, _VerifyErrors):
1414
  """Verifies the cluster config.
1415

1416
  """
1417
  REQ_BGL = True
1418

    
1419
  def _VerifyHVP(self, hvp_data):
1420
    """Verifies locally the syntax of the hypervisor parameters.
1421

1422
    """
1423
    for item, hv_name, hv_params in hvp_data:
1424
      msg = ("hypervisor %s parameters syntax check (source %s): %%s" %
1425
             (item, hv_name))
1426
      try:
1427
        hv_class = hypervisor.GetHypervisor(hv_name)
1428
        utils.ForceDictType(hv_params, constants.HVS_PARAMETER_TYPES)
1429
        hv_class.CheckParameterSyntax(hv_params)
1430
      except errors.GenericError, err:
1431
        self._ErrorIf(True, self.ECLUSTERCFG, None, msg % str(err))
1432

    
1433
  def ExpandNames(self):
1434
    # Information can be safely retrieved as the BGL is acquired in exclusive
1435
    # mode
1436
    self.all_group_info = self.cfg.GetAllNodeGroupsInfo()
1437
    self.all_node_info = self.cfg.GetAllNodesInfo()
1438
    self.all_inst_info = self.cfg.GetAllInstancesInfo()
1439
    self.needed_locks = {}
1440

    
1441
  def Exec(self, feedback_fn):
1442
    """Verify integrity of cluster, performing various test on nodes.
1443

1444
    """
1445
    self.bad = False
1446
    self._feedback_fn = feedback_fn
1447

    
1448
    feedback_fn("* Verifying cluster config")
1449

    
1450
    for msg in self.cfg.VerifyConfig():
1451
      self._ErrorIf(True, self.ECLUSTERCFG, None, msg)
1452

    
1453
    feedback_fn("* Verifying cluster certificate files")
1454

    
1455
    for cert_filename in constants.ALL_CERT_FILES:
1456
      (errcode, msg) = _VerifyCertificate(cert_filename)
1457
      self._ErrorIf(errcode, self.ECLUSTERCERT, None, msg, code=errcode)
1458

    
1459
    feedback_fn("* Verifying hypervisor parameters")
1460

    
1461
    self._VerifyHVP(_GetAllHypervisorParameters(self.cfg.GetClusterInfo(),
1462
                                                self.all_inst_info.values()))
1463

    
1464
    feedback_fn("* Verifying all nodes belong to an existing group")
1465

    
1466
    # We do this verification here because, should this bogus circumstance
1467
    # occur, it would never be caught by VerifyGroup, which only acts on
1468
    # nodes/instances reachable from existing node groups.
1469

    
1470
    dangling_nodes = set(node.name for node in self.all_node_info.values()
1471
                         if node.group not in self.all_group_info)
1472

    
1473
    dangling_instances = {}
1474
    no_node_instances = []
1475

    
1476
    for inst in self.all_inst_info.values():
1477
      if inst.primary_node in dangling_nodes:
1478
        dangling_instances.setdefault(inst.primary_node, []).append(inst.name)
1479
      elif inst.primary_node not in self.all_node_info:
1480
        no_node_instances.append(inst.name)
1481

    
1482
    pretty_dangling = [
1483
        "%s (%s)" %
1484
        (node.name,
1485
         utils.CommaJoin(dangling_instances.get(node.name,
1486
                                                ["no instances"])))
1487
        for node in dangling_nodes]
1488

    
1489
    self._ErrorIf(bool(dangling_nodes), self.ECLUSTERDANGLINGNODES, None,
1490
                  "the following nodes (and their instances) belong to a non"
1491
                  " existing group: %s", utils.CommaJoin(pretty_dangling))
1492

    
1493
    self._ErrorIf(bool(no_node_instances), self.ECLUSTERDANGLINGINST, None,
1494
                  "the following instances have a non-existing primary-node:"
1495
                  " %s", utils.CommaJoin(no_node_instances))
1496

    
1497
    return (not self.bad, [g.name for g in self.all_group_info.values()])
1498

    
1499

    
1500
class LUClusterVerifyGroup(LogicalUnit, _VerifyErrors):
1501
  """Verifies the status of a node group.
1502

1503
  """
1504
  HPATH = "cluster-verify"
1505
  HTYPE = constants.HTYPE_CLUSTER
1506
  REQ_BGL = False
1507

    
1508
  _HOOKS_INDENT_RE = re.compile("^", re.M)
1509

    
1510
  class NodeImage(object):
1511
    """A class representing the logical and physical status of a node.
1512

1513
    @type name: string
1514
    @ivar name: the node name to which this object refers
1515
    @ivar volumes: a structure as returned from
1516
        L{ganeti.backend.GetVolumeList} (runtime)
1517
    @ivar instances: a list of running instances (runtime)
1518
    @ivar pinst: list of configured primary instances (config)
1519
    @ivar sinst: list of configured secondary instances (config)
1520
    @ivar sbp: dictionary of {primary-node: list of instances} for all
1521
        instances for which this node is secondary (config)
1522
    @ivar mfree: free memory, as reported by hypervisor (runtime)
1523
    @ivar dfree: free disk, as reported by the node (runtime)
1524
    @ivar offline: the offline status (config)
1525
    @type rpc_fail: boolean
1526
    @ivar rpc_fail: whether the RPC verify call was successfull (overall,
1527
        not whether the individual keys were correct) (runtime)
1528
    @type lvm_fail: boolean
1529
    @ivar lvm_fail: whether the RPC call didn't return valid LVM data
1530
    @type hyp_fail: boolean
1531
    @ivar hyp_fail: whether the RPC call didn't return the instance list
1532
    @type ghost: boolean
1533
    @ivar ghost: whether this is a known node or not (config)
1534
    @type os_fail: boolean
1535
    @ivar os_fail: whether the RPC call didn't return valid OS data
1536
    @type oslist: list
1537
    @ivar oslist: list of OSes as diagnosed by DiagnoseOS
1538
    @type vm_capable: boolean
1539
    @ivar vm_capable: whether the node can host instances
1540

1541
    """
1542
    def __init__(self, offline=False, name=None, vm_capable=True):
1543
      self.name = name
1544
      self.volumes = {}
1545
      self.instances = []
1546
      self.pinst = []
1547
      self.sinst = []
1548
      self.sbp = {}
1549
      self.mfree = 0
1550
      self.dfree = 0
1551
      self.offline = offline
1552
      self.vm_capable = vm_capable
1553
      self.rpc_fail = False
1554
      self.lvm_fail = False
1555
      self.hyp_fail = False
1556
      self.ghost = False
1557
      self.os_fail = False
1558
      self.oslist = {}
1559

    
1560
  def ExpandNames(self):
1561
    # This raises errors.OpPrereqError on its own:
1562
    self.group_uuid = self.cfg.LookupNodeGroup(self.op.group_name)
1563

    
1564
    # Get instances in node group; this is unsafe and needs verification later
1565
    inst_names = self.cfg.GetNodeGroupInstances(self.group_uuid)
1566

    
1567
    self.needed_locks = {
1568
      locking.LEVEL_INSTANCE: inst_names,
1569
      locking.LEVEL_NODEGROUP: [self.group_uuid],
1570
      locking.LEVEL_NODE: [],
1571
      }
1572

    
1573
    self.share_locks = dict.fromkeys(locking.LEVELS, 1)
1574

    
1575
  def DeclareLocks(self, level):
1576
    if level == locking.LEVEL_NODE:
1577
      # Get members of node group; this is unsafe and needs verification later
1578
      nodes = set(self.cfg.GetNodeGroup(self.group_uuid).members)
1579

    
1580
      all_inst_info = self.cfg.GetAllInstancesInfo()
1581

    
1582
      # In Exec(), we warn about mirrored instances that have primary and
1583
      # secondary living in separate node groups. To fully verify that
1584
      # volumes for these instances are healthy, we will need to do an
1585
      # extra call to their secondaries. We ensure here those nodes will
1586
      # be locked.
1587
      for inst in self.glm.list_owned(locking.LEVEL_INSTANCE):
1588
        # Important: access only the instances whose lock is owned
1589
        if all_inst_info[inst].disk_template in constants.DTS_INT_MIRROR:
1590
          nodes.update(all_inst_info[inst].secondary_nodes)
1591

    
1592
      self.needed_locks[locking.LEVEL_NODE] = nodes
1593

    
1594
  def CheckPrereq(self):
1595
    group_nodes = set(self.cfg.GetNodeGroup(self.group_uuid).members)
1596
    group_instances = self.cfg.GetNodeGroupInstances(self.group_uuid)
1597

    
1598
    unlocked_nodes = \
1599
        group_nodes.difference(self.glm.list_owned(locking.LEVEL_NODE))
1600

    
1601
    unlocked_instances = \
1602
        group_instances.difference(self.glm.list_owned(locking.LEVEL_INSTANCE))
1603

    
1604
    if unlocked_nodes:
1605
      raise errors.OpPrereqError("Missing lock for nodes: %s" %
1606
                                 utils.CommaJoin(unlocked_nodes))
1607

    
1608
    if unlocked_instances:
1609
      raise errors.OpPrereqError("Missing lock for instances: %s" %
1610
                                 utils.CommaJoin(unlocked_instances))
1611

    
1612
    self.all_node_info = self.cfg.GetAllNodesInfo()
1613
    self.all_inst_info = self.cfg.GetAllInstancesInfo()
1614

    
1615
    self.my_node_names = utils.NiceSort(group_nodes)
1616
    self.my_inst_names = utils.NiceSort(group_instances)
1617

    
1618
    self.my_node_info = dict((name, self.all_node_info[name])
1619
                             for name in self.my_node_names)
1620

    
1621
    self.my_inst_info = dict((name, self.all_inst_info[name])
1622
                             for name in self.my_inst_names)
1623

    
1624
    # We detect here the nodes that will need the extra RPC calls for verifying
1625
    # split LV volumes; they should be locked.
1626
    extra_lv_nodes = set()
1627

    
1628
    for inst in self.my_inst_info.values():
1629
      if inst.disk_template in constants.DTS_INT_MIRROR:
1630
        group = self.my_node_info[inst.primary_node].group
1631
        for nname in inst.secondary_nodes:
1632
          if self.all_node_info[nname].group != group:
1633
            extra_lv_nodes.add(nname)
1634

    
1635
    unlocked_lv_nodes = \
1636
        extra_lv_nodes.difference(self.glm.list_owned(locking.LEVEL_NODE))
1637

    
1638
    if unlocked_lv_nodes:
1639
      raise errors.OpPrereqError("these nodes could be locked: %s" %
1640
                                 utils.CommaJoin(unlocked_lv_nodes))
1641
    self.extra_lv_nodes = list(extra_lv_nodes)
1642

    
1643
  def _VerifyNode(self, ninfo, nresult):
1644
    """Perform some basic validation on data returned from a node.
1645

1646
      - check the result data structure is well formed and has all the
1647
        mandatory fields
1648
      - check ganeti version
1649

1650
    @type ninfo: L{objects.Node}
1651
    @param ninfo: the node to check
1652
    @param nresult: the results from the node
1653
    @rtype: boolean
1654
    @return: whether overall this call was successful (and we can expect
1655
         reasonable values in the respose)
1656

1657
    """
1658
    node = ninfo.name
1659
    _ErrorIf = self._ErrorIf # pylint: disable-msg=C0103
1660

    
1661
    # main result, nresult should be a non-empty dict
1662
    test = not nresult or not isinstance(nresult, dict)
1663
    _ErrorIf(test, self.ENODERPC, node,
1664
                  "unable to verify node: no data returned")
1665
    if test:
1666
      return False
1667

    
1668
    # compares ganeti version
1669
    local_version = constants.PROTOCOL_VERSION
1670
    remote_version = nresult.get("version", None)
1671
    test = not (remote_version and
1672
                isinstance(remote_version, (list, tuple)) and
1673
                len(remote_version) == 2)
1674
    _ErrorIf(test, self.ENODERPC, node,
1675
             "connection to node returned invalid data")
1676
    if test:
1677
      return False
1678

    
1679
    test = local_version != remote_version[0]
1680
    _ErrorIf(test, self.ENODEVERSION, node,
1681
             "incompatible protocol versions: master %s,"
1682
             " node %s", local_version, remote_version[0])
1683
    if test:
1684
      return False
1685

    
1686
    # node seems compatible, we can actually try to look into its results
1687

    
1688
    # full package version
1689
    self._ErrorIf(constants.RELEASE_VERSION != remote_version[1],
1690
                  self.ENODEVERSION, node,
1691
                  "software version mismatch: master %s, node %s",
1692
                  constants.RELEASE_VERSION, remote_version[1],
1693
                  code=self.ETYPE_WARNING)
1694

    
1695
    hyp_result = nresult.get(constants.NV_HYPERVISOR, None)
1696
    if ninfo.vm_capable and isinstance(hyp_result, dict):
1697
      for hv_name, hv_result in hyp_result.iteritems():
1698
        test = hv_result is not None
1699
        _ErrorIf(test, self.ENODEHV, node,
1700
                 "hypervisor %s verify failure: '%s'", hv_name, hv_result)
1701

    
1702
    hvp_result = nresult.get(constants.NV_HVPARAMS, None)
1703
    if ninfo.vm_capable and isinstance(hvp_result, list):
1704
      for item, hv_name, hv_result in hvp_result:
1705
        _ErrorIf(True, self.ENODEHV, node,
1706
                 "hypervisor %s parameter verify failure (source %s): %s",
1707
                 hv_name, item, hv_result)
1708

    
1709
    test = nresult.get(constants.NV_NODESETUP,
1710
                       ["Missing NODESETUP results"])
1711
    _ErrorIf(test, self.ENODESETUP, node, "node setup error: %s",
1712
             "; ".join(test))
1713

    
1714
    return True
1715

    
1716
  def _VerifyNodeTime(self, ninfo, nresult,
1717
                      nvinfo_starttime, nvinfo_endtime):
1718
    """Check the node time.
1719

1720
    @type ninfo: L{objects.Node}
1721
    @param ninfo: the node to check
1722
    @param nresult: the remote results for the node
1723
    @param nvinfo_starttime: the start time of the RPC call
1724
    @param nvinfo_endtime: the end time of the RPC call
1725

1726
    """
1727
    node = ninfo.name
1728
    _ErrorIf = self._ErrorIf # pylint: disable-msg=C0103
1729

    
1730
    ntime = nresult.get(constants.NV_TIME, None)
1731
    try:
1732
      ntime_merged = utils.MergeTime(ntime)
1733
    except (ValueError, TypeError):
1734
      _ErrorIf(True, self.ENODETIME, node, "Node returned invalid time")
1735
      return
1736

    
1737
    if ntime_merged < (nvinfo_starttime - constants.NODE_MAX_CLOCK_SKEW):
1738
      ntime_diff = "%.01fs" % abs(nvinfo_starttime - ntime_merged)
1739
    elif ntime_merged > (nvinfo_endtime + constants.NODE_MAX_CLOCK_SKEW):
1740
      ntime_diff = "%.01fs" % abs(ntime_merged - nvinfo_endtime)
1741
    else:
1742
      ntime_diff = None
1743

    
1744
    _ErrorIf(ntime_diff is not None, self.ENODETIME, node,
1745
             "Node time diverges by at least %s from master node time",
1746
             ntime_diff)
1747

    
1748
  def _VerifyNodeLVM(self, ninfo, nresult, vg_name):
1749
    """Check the node LVM results.
1750

1751
    @type ninfo: L{objects.Node}
1752
    @param ninfo: the node to check
1753
    @param nresult: the remote results for the node
1754
    @param vg_name: the configured VG name
1755

1756
    """
1757
    if vg_name is None:
1758
      return
1759

    
1760
    node = ninfo.name
1761
    _ErrorIf = self._ErrorIf # pylint: disable-msg=C0103
1762

    
1763
    # checks vg existence and size > 20G
1764
    vglist = nresult.get(constants.NV_VGLIST, None)
1765
    test = not vglist
1766
    _ErrorIf(test, self.ENODELVM, node, "unable to check volume groups")
1767
    if not test:
1768
      vgstatus = utils.CheckVolumeGroupSize(vglist, vg_name,
1769
                                            constants.MIN_VG_SIZE)
1770
      _ErrorIf(vgstatus, self.ENODELVM, node, vgstatus)
1771

    
1772
    # check pv names
1773
    pvlist = nresult.get(constants.NV_PVLIST, None)
1774
    test = pvlist is None
1775
    _ErrorIf(test, self.ENODELVM, node, "Can't get PV list from node")
1776
    if not test:
1777
      # check that ':' is not present in PV names, since it's a
1778
      # special character for lvcreate (denotes the range of PEs to
1779
      # use on the PV)
1780
      for _, pvname, owner_vg in pvlist:
1781
        test = ":" in pvname
1782
        _ErrorIf(test, self.ENODELVM, node, "Invalid character ':' in PV"
1783
                 " '%s' of VG '%s'", pvname, owner_vg)
1784

    
1785
  def _VerifyNodeBridges(self, ninfo, nresult, bridges):
1786
    """Check the node bridges.
1787

1788
    @type ninfo: L{objects.Node}
1789
    @param ninfo: the node to check
1790
    @param nresult: the remote results for the node
1791
    @param bridges: the expected list of bridges
1792

1793
    """
1794
    if not bridges:
1795
      return
1796

    
1797
    node = ninfo.name
1798
    _ErrorIf = self._ErrorIf # pylint: disable-msg=C0103
1799

    
1800
    missing = nresult.get(constants.NV_BRIDGES, None)
1801
    test = not isinstance(missing, list)
1802
    _ErrorIf(test, self.ENODENET, node,
1803
             "did not return valid bridge information")
1804
    if not test:
1805
      _ErrorIf(bool(missing), self.ENODENET, node, "missing bridges: %s" %
1806
               utils.CommaJoin(sorted(missing)))
1807

    
1808
  def _VerifyNodeNetwork(self, ninfo, nresult):
1809
    """Check the node network connectivity results.
1810

1811
    @type ninfo: L{objects.Node}
1812
    @param ninfo: the node to check
1813
    @param nresult: the remote results for the node
1814

1815
    """
1816
    node = ninfo.name
1817
    _ErrorIf = self._ErrorIf # pylint: disable-msg=C0103
1818

    
1819
    test = constants.NV_NODELIST not in nresult
1820
    _ErrorIf(test, self.ENODESSH, node,
1821
             "node hasn't returned node ssh connectivity data")
1822
    if not test:
1823
      if nresult[constants.NV_NODELIST]:
1824
        for a_node, a_msg in nresult[constants.NV_NODELIST].items():
1825
          _ErrorIf(True, self.ENODESSH, node,
1826
                   "ssh communication with node '%s': %s", a_node, a_msg)
1827

    
1828
    test = constants.NV_NODENETTEST not in nresult
1829
    _ErrorIf(test, self.ENODENET, node,
1830
             "node hasn't returned node tcp connectivity data")
1831
    if not test:
1832
      if nresult[constants.NV_NODENETTEST]:
1833
        nlist = utils.NiceSort(nresult[constants.NV_NODENETTEST].keys())
1834
        for anode in nlist:
1835
          _ErrorIf(True, self.ENODENET, node,
1836
                   "tcp communication with node '%s': %s",
1837
                   anode, nresult[constants.NV_NODENETTEST][anode])
1838

    
1839
    test = constants.NV_MASTERIP not in nresult
1840
    _ErrorIf(test, self.ENODENET, node,
1841
             "node hasn't returned node master IP reachability data")
1842
    if not test:
1843
      if not nresult[constants.NV_MASTERIP]:
1844
        if node == self.master_node:
1845
          msg = "the master node cannot reach the master IP (not configured?)"
1846
        else:
1847
          msg = "cannot reach the master IP"
1848
        _ErrorIf(True, self.ENODENET, node, msg)
1849

    
1850
  def _VerifyInstance(self, instance, instanceconfig, node_image,
1851
                      diskstatus):
1852
    """Verify an instance.
1853

1854
    This function checks to see if the required block devices are
1855
    available on the instance's node.
1856

1857
    """
1858
    _ErrorIf = self._ErrorIf # pylint: disable-msg=C0103
1859
    node_current = instanceconfig.primary_node
1860

    
1861
    node_vol_should = {}
1862
    instanceconfig.MapLVsByNode(node_vol_should)
1863

    
1864
    for node in node_vol_should:
1865
      n_img = node_image[node]
1866
      if n_img.offline or n_img.rpc_fail or n_img.lvm_fail:
1867
        # ignore missing volumes on offline or broken nodes
1868
        continue
1869
      for volume in node_vol_should[node]:
1870
        test = volume not in n_img.volumes
1871
        _ErrorIf(test, self.EINSTANCEMISSINGDISK, instance,
1872
                 "volume %s missing on node %s", volume, node)
1873

    
1874
    if instanceconfig.admin_up:
1875
      pri_img = node_image[node_current]
1876
      test = instance not in pri_img.instances and not pri_img.offline
1877
      _ErrorIf(test, self.EINSTANCEDOWN, instance,
1878
               "instance not running on its primary node %s",
1879
               node_current)
1880

    
1881
    diskdata = [(nname, success, status, idx)
1882
                for (nname, disks) in diskstatus.items()
1883
                for idx, (success, status) in enumerate(disks)]
1884

    
1885
    for nname, success, bdev_status, idx in diskdata:
1886
      # the 'ghost node' construction in Exec() ensures that we have a
1887
      # node here
1888
      snode = node_image[nname]
1889
      bad_snode = snode.ghost or snode.offline
1890
      _ErrorIf(instanceconfig.admin_up and not success and not bad_snode,
1891
               self.EINSTANCEFAULTYDISK, instance,
1892
               "couldn't retrieve status for disk/%s on %s: %s",
1893
               idx, nname, bdev_status)
1894
      _ErrorIf((instanceconfig.admin_up and success and
1895
                bdev_status.ldisk_status == constants.LDS_FAULTY),
1896
               self.EINSTANCEFAULTYDISK, instance,
1897
               "disk/%s on %s is faulty", idx, nname)
1898

    
1899
  def _VerifyOrphanVolumes(self, node_vol_should, node_image, reserved):
1900
    """Verify if there are any unknown volumes in the cluster.
1901

1902
    The .os, .swap and backup volumes are ignored. All other volumes are
1903
    reported as unknown.
1904

1905
    @type reserved: L{ganeti.utils.FieldSet}
1906
    @param reserved: a FieldSet of reserved volume names
1907

1908
    """
1909
    for node, n_img in node_image.items():
1910
      if n_img.offline or n_img.rpc_fail or n_img.lvm_fail:
1911
        # skip non-healthy nodes
1912
        continue
1913
      for volume in n_img.volumes:
1914
        test = ((node not in node_vol_should or
1915
                volume not in node_vol_should[node]) and
1916
                not reserved.Matches(volume))
1917
        self._ErrorIf(test, self.ENODEORPHANLV, node,
1918
                      "volume %s is unknown", volume)
1919

    
1920
  def _VerifyNPlusOneMemory(self, node_image, instance_cfg):
1921
    """Verify N+1 Memory Resilience.
1922

1923
    Check that if one single node dies we can still start all the
1924
    instances it was primary for.
1925

1926
    """
1927
    cluster_info = self.cfg.GetClusterInfo()
1928
    for node, n_img in node_image.items():
1929
      # This code checks that every node which is now listed as
1930
      # secondary has enough memory to host all instances it is
1931
      # supposed to should a single other node in the cluster fail.
1932
      # FIXME: not ready for failover to an arbitrary node
1933
      # FIXME: does not support file-backed instances
1934
      # WARNING: we currently take into account down instances as well
1935
      # as up ones, considering that even if they're down someone
1936
      # might want to start them even in the event of a node failure.
1937
      if n_img.offline:
1938
        # we're skipping offline nodes from the N+1 warning, since
1939
        # most likely we don't have good memory infromation from them;
1940
        # we already list instances living on such nodes, and that's
1941
        # enough warning
1942
        continue
1943
      for prinode, instances in n_img.sbp.items():
1944
        needed_mem = 0
1945
        for instance in instances:
1946
          bep = cluster_info.FillBE(instance_cfg[instance])
1947
          if bep[constants.BE_AUTO_BALANCE]:
1948
            needed_mem += bep[constants.BE_MEMORY]
1949
        test = n_img.mfree < needed_mem
1950
        self._ErrorIf(test, self.ENODEN1, node,
1951
                      "not enough memory to accomodate instance failovers"
1952
                      " should node %s fail (%dMiB needed, %dMiB available)",
1953
                      prinode, needed_mem, n_img.mfree)
1954

    
1955
  @classmethod
1956
  def _VerifyFiles(cls, errorif, nodeinfo, master_node, all_nvinfo,
1957
                   (files_all, files_all_opt, files_mc, files_vm)):
1958
    """Verifies file checksums collected from all nodes.
1959

1960
    @param errorif: Callback for reporting errors
1961
    @param nodeinfo: List of L{objects.Node} objects
1962
    @param master_node: Name of master node
1963
    @param all_nvinfo: RPC results
1964

1965
    """
1966
    node_names = frozenset(node.name for node in nodeinfo)
1967

    
1968
    assert master_node in node_names
1969
    assert (len(files_all | files_all_opt | files_mc | files_vm) ==
1970
            sum(map(len, [files_all, files_all_opt, files_mc, files_vm]))), \
1971
           "Found file listed in more than one file list"
1972

    
1973
    # Define functions determining which nodes to consider for a file
1974
    file2nodefn = dict([(filename, fn)
1975
      for (files, fn) in [(files_all, None),
1976
                          (files_all_opt, None),
1977
                          (files_mc, lambda node: (node.master_candidate or
1978
                                                   node.name == master_node)),
1979
                          (files_vm, lambda node: node.vm_capable)]
1980
      for filename in files])
1981

    
1982
    fileinfo = dict((filename, {}) for filename in file2nodefn.keys())
1983

    
1984
    for node in nodeinfo:
1985
      nresult = all_nvinfo[node.name]
1986

    
1987
      if nresult.fail_msg or not nresult.payload:
1988
        node_files = None
1989
      else:
1990
        node_files = nresult.payload.get(constants.NV_FILELIST, None)
1991

    
1992
      test = not (node_files and isinstance(node_files, dict))
1993
      errorif(test, cls.ENODEFILECHECK, node.name,
1994
              "Node did not return file checksum data")
1995
      if test:
1996
        continue
1997

    
1998
      for (filename, checksum) in node_files.items():
1999
        # Check if the file should be considered for a node
2000
        fn = file2nodefn[filename]
2001
        if fn is None or fn(node):
2002
          fileinfo[filename].setdefault(checksum, set()).add(node.name)
2003

    
2004
    for (filename, checksums) in fileinfo.items():
2005
      assert compat.all(len(i) > 10 for i in checksums), "Invalid checksum"
2006

    
2007
      # Nodes having the file
2008
      with_file = frozenset(node_name
2009
                            for nodes in fileinfo[filename].values()
2010
                            for node_name in nodes)
2011

    
2012
      # Nodes missing file
2013
      missing_file = node_names - with_file
2014

    
2015
      if filename in files_all_opt:
2016
        # All or no nodes
2017
        errorif(missing_file and missing_file != node_names,
2018
                cls.ECLUSTERFILECHECK, None,
2019
                "File %s is optional, but it must exist on all or no"
2020
                " nodes (not found on %s)",
2021
                filename, utils.CommaJoin(utils.NiceSort(missing_file)))
2022
      else:
2023
        errorif(missing_file, cls.ECLUSTERFILECHECK, None,
2024
                "File %s is missing from node(s) %s", filename,
2025
                utils.CommaJoin(utils.NiceSort(missing_file)))
2026

    
2027
      # See if there are multiple versions of the file
2028
      test = len(checksums) > 1
2029
      if test:
2030
        variants = ["variant %s on %s" %
2031
                    (idx + 1, utils.CommaJoin(utils.NiceSort(nodes)))
2032
                    for (idx, (checksum, nodes)) in
2033
                      enumerate(sorted(checksums.items()))]
2034
      else:
2035
        variants = []
2036

    
2037
      errorif(test, cls.ECLUSTERFILECHECK, None,
2038
              "File %s found with %s different checksums (%s)",
2039
              filename, len(checksums), "; ".join(variants))
2040

    
2041
  def _VerifyNodeDrbd(self, ninfo, nresult, instanceinfo, drbd_helper,
2042
                      drbd_map):
2043
    """Verifies and the node DRBD status.
2044

2045
    @type ninfo: L{objects.Node}
2046
    @param ninfo: the node to check
2047
    @param nresult: the remote results for the node
2048
    @param instanceinfo: the dict of instances
2049
    @param drbd_helper: the configured DRBD usermode helper
2050
    @param drbd_map: the DRBD map as returned by
2051
        L{ganeti.config.ConfigWriter.ComputeDRBDMap}
2052

2053
    """
2054
    node = ninfo.name
2055
    _ErrorIf = self._ErrorIf # pylint: disable-msg=C0103
2056

    
2057
    if drbd_helper:
2058
      helper_result = nresult.get(constants.NV_DRBDHELPER, None)
2059
      test = (helper_result == None)
2060
      _ErrorIf(test, self.ENODEDRBDHELPER, node,
2061
               "no drbd usermode helper returned")
2062
      if helper_result:
2063
        status, payload = helper_result
2064
        test = not status
2065
        _ErrorIf(test, self.ENODEDRBDHELPER, node,
2066
                 "drbd usermode helper check unsuccessful: %s", payload)
2067
        test = status and (payload != drbd_helper)
2068
        _ErrorIf(test, self.ENODEDRBDHELPER, node,
2069
                 "wrong drbd usermode helper: %s", payload)
2070

    
2071
    # compute the DRBD minors
2072
    node_drbd = {}
2073
    for minor, instance in drbd_map[node].items():
2074
      test = instance not in instanceinfo
2075
      _ErrorIf(test, self.ECLUSTERCFG, None,
2076
               "ghost instance '%s' in temporary DRBD map", instance)
2077
        # ghost instance should not be running, but otherwise we
2078
        # don't give double warnings (both ghost instance and
2079
        # unallocated minor in use)
2080
      if test:
2081
        node_drbd[minor] = (instance, False)
2082
      else:
2083
        instance = instanceinfo[instance]
2084
        node_drbd[minor] = (instance.name, instance.admin_up)
2085

    
2086
    # and now check them
2087
    used_minors = nresult.get(constants.NV_DRBDLIST, [])
2088
    test = not isinstance(used_minors, (tuple, list))
2089
    _ErrorIf(test, self.ENODEDRBD, node,
2090
             "cannot parse drbd status file: %s", str(used_minors))
2091
    if test:
2092
      # we cannot check drbd status
2093
      return
2094

    
2095
    for minor, (iname, must_exist) in node_drbd.items():
2096
      test = minor not in used_minors and must_exist
2097
      _ErrorIf(test, self.ENODEDRBD, node,
2098
               "drbd minor %d of instance %s is not active", minor, iname)
2099
    for minor in used_minors:
2100
      test = minor not in node_drbd
2101
      _ErrorIf(test, self.ENODEDRBD, node,
2102
               "unallocated drbd minor %d is in use", minor)
2103

    
2104
  def _UpdateNodeOS(self, ninfo, nresult, nimg):
2105
    """Builds the node OS structures.
2106

2107
    @type ninfo: L{objects.Node}
2108
    @param ninfo: the node to check
2109
    @param nresult: the remote results for the node
2110
    @param nimg: the node image object
2111

2112
    """
2113
    node = ninfo.name
2114
    _ErrorIf = self._ErrorIf # pylint: disable-msg=C0103
2115

    
2116
    remote_os = nresult.get(constants.NV_OSLIST, None)
2117
    test = (not isinstance(remote_os, list) or
2118
            not compat.all(isinstance(v, list) and len(v) == 7
2119
                           for v in remote_os))
2120

    
2121
    _ErrorIf(test, self.ENODEOS, node,
2122
             "node hasn't returned valid OS data")
2123

    
2124
    nimg.os_fail = test
2125

    
2126
    if test:
2127
      return
2128

    
2129
    os_dict = {}
2130

    
2131
    for (name, os_path, status, diagnose,
2132
         variants, parameters, api_ver) in nresult[constants.NV_OSLIST]:
2133

    
2134
      if name not in os_dict:
2135
        os_dict[name] = []
2136

    
2137
      # parameters is a list of lists instead of list of tuples due to
2138
      # JSON lacking a real tuple type, fix it:
2139
      parameters = [tuple(v) for v in parameters]
2140
      os_dict[name].append((os_path, status, diagnose,
2141
                            set(variants), set(parameters), set(api_ver)))
2142

    
2143
    nimg.oslist = os_dict
2144

    
2145
  def _VerifyNodeOS(self, ninfo, nimg, base):
2146
    """Verifies the node OS list.
2147

2148
    @type ninfo: L{objects.Node}
2149
    @param ninfo: the node to check
2150
    @param nimg: the node image object
2151
    @param base: the 'template' node we match against (e.g. from the master)
2152

2153
    """
2154
    node = ninfo.name
2155
    _ErrorIf = self._ErrorIf # pylint: disable-msg=C0103
2156

    
2157
    assert not nimg.os_fail, "Entered _VerifyNodeOS with failed OS rpc?"
2158

    
2159
    beautify_params = lambda l: ["%s: %s" % (k, v) for (k, v) in l]
2160
    for os_name, os_data in nimg.oslist.items():
2161
      assert os_data, "Empty OS status for OS %s?!" % os_name
2162
      f_path, f_status, f_diag, f_var, f_param, f_api = os_data[0]
2163
      _ErrorIf(not f_status, self.ENODEOS, node,
2164
               "Invalid OS %s (located at %s): %s", os_name, f_path, f_diag)
2165
      _ErrorIf(len(os_data) > 1, self.ENODEOS, node,
2166
               "OS '%s' has multiple entries (first one shadows the rest): %s",
2167
               os_name, utils.CommaJoin([v[0] for v in os_data]))
2168
      # this will catched in backend too
2169
      _ErrorIf(compat.any(v >= constants.OS_API_V15 for v in f_api)
2170
               and not f_var, self.ENODEOS, node,
2171
               "OS %s with API at least %d does not declare any variant",
2172
               os_name, constants.OS_API_V15)
2173
      # comparisons with the 'base' image
2174
      test = os_name not in base.oslist
2175
      _ErrorIf(test, self.ENODEOS, node,
2176
               "Extra OS %s not present on reference node (%s)",
2177
               os_name, base.name)
2178
      if test:
2179
        continue
2180
      assert base.oslist[os_name], "Base node has empty OS status?"
2181
      _, b_status, _, b_var, b_param, b_api = base.oslist[os_name][0]
2182
      if not b_status:
2183
        # base OS is invalid, skipping
2184
        continue
2185
      for kind, a, b in [("API version", f_api, b_api),
2186
                         ("variants list", f_var, b_var),
2187
                         ("parameters", beautify_params(f_param),
2188
                          beautify_params(b_param))]:
2189
        _ErrorIf(a != b, self.ENODEOS, node,
2190
                 "OS %s for %s differs from reference node %s: [%s] vs. [%s]",
2191
                 kind, os_name, base.name,
2192
                 utils.CommaJoin(sorted(a)), utils.CommaJoin(sorted(b)))
2193

    
2194
    # check any missing OSes
2195
    missing = set(base.oslist.keys()).difference(nimg.oslist.keys())
2196
    _ErrorIf(missing, self.ENODEOS, node,
2197
             "OSes present on reference node %s but missing on this node: %s",
2198
             base.name, utils.CommaJoin(missing))
2199

    
2200
  def _VerifyOob(self, ninfo, nresult):
2201
    """Verifies out of band functionality of a node.
2202

2203
    @type ninfo: L{objects.Node}
2204
    @param ninfo: the node to check
2205
    @param nresult: the remote results for the node
2206

2207
    """
2208
    node = ninfo.name
2209
    # We just have to verify the paths on master and/or master candidates
2210
    # as the oob helper is invoked on the master
2211
    if ((ninfo.master_candidate or ninfo.master_capable) and
2212
        constants.NV_OOB_PATHS in nresult):
2213
      for path_result in nresult[constants.NV_OOB_PATHS]:
2214
        self._ErrorIf(path_result, self.ENODEOOBPATH, node, path_result)
2215

    
2216
  def _UpdateNodeVolumes(self, ninfo, nresult, nimg, vg_name):
2217
    """Verifies and updates the node volume data.
2218

2219
    This function will update a L{NodeImage}'s internal structures
2220
    with data from the remote call.
2221

2222
    @type ninfo: L{objects.Node}
2223
    @param ninfo: the node to check
2224
    @param nresult: the remote results for the node
2225
    @param nimg: the node image object
2226
    @param vg_name: the configured VG name
2227

2228
    """
2229
    node = ninfo.name
2230
    _ErrorIf = self._ErrorIf # pylint: disable-msg=C0103
2231

    
2232
    nimg.lvm_fail = True
2233
    lvdata = nresult.get(constants.NV_LVLIST, "Missing LV data")
2234
    if vg_name is None:
2235
      pass
2236
    elif isinstance(lvdata, basestring):
2237
      _ErrorIf(True, self.ENODELVM, node, "LVM problem on node: %s",
2238
               utils.SafeEncode(lvdata))
2239
    elif not isinstance(lvdata, dict):
2240
      _ErrorIf(True, self.ENODELVM, node, "rpc call to node failed (lvlist)")
2241
    else:
2242
      nimg.volumes = lvdata
2243
      nimg.lvm_fail = False
2244

    
2245
  def _UpdateNodeInstances(self, ninfo, nresult, nimg):
2246
    """Verifies and updates the node instance list.
2247

2248
    If the listing was successful, then updates this node's instance
2249
    list. Otherwise, it marks the RPC call as failed for the instance
2250
    list key.
2251

2252
    @type ninfo: L{objects.Node}
2253
    @param ninfo: the node to check
2254
    @param nresult: the remote results for the node
2255
    @param nimg: the node image object
2256

2257
    """
2258
    idata = nresult.get(constants.NV_INSTANCELIST, None)
2259
    test = not isinstance(idata, list)
2260
    self._ErrorIf(test, self.ENODEHV, ninfo.name, "rpc call to node failed"
2261
                  " (instancelist): %s", utils.SafeEncode(str(idata)))
2262
    if test:
2263
      nimg.hyp_fail = True
2264
    else:
2265
      nimg.instances = idata
2266

    
2267
  def _UpdateNodeInfo(self, ninfo, nresult, nimg, vg_name):
2268
    """Verifies and computes a node information map
2269

2270
    @type ninfo: L{objects.Node}
2271
    @param ninfo: the node to check
2272
    @param nresult: the remote results for the node
2273
    @param nimg: the node image object
2274
    @param vg_name: the configured VG name
2275

2276
    """
2277
    node = ninfo.name
2278
    _ErrorIf = self._ErrorIf # pylint: disable-msg=C0103
2279

    
2280
    # try to read free memory (from the hypervisor)
2281
    hv_info = nresult.get(constants.NV_HVINFO, None)
2282
    test = not isinstance(hv_info, dict) or "memory_free" not in hv_info
2283
    _ErrorIf(test, self.ENODEHV, node, "rpc call to node failed (hvinfo)")
2284
    if not test:
2285
      try:
2286
        nimg.mfree = int(hv_info["memory_free"])
2287
      except (ValueError, TypeError):
2288
        _ErrorIf(True, self.ENODERPC, node,
2289
                 "node returned invalid nodeinfo, check hypervisor")
2290

    
2291
    # FIXME: devise a free space model for file based instances as well
2292
    if vg_name is not None:
2293
      test = (constants.NV_VGLIST not in nresult or
2294
              vg_name not in nresult[constants.NV_VGLIST])
2295
      _ErrorIf(test, self.ENODELVM, node,
2296
               "node didn't return data for the volume group '%s'"
2297
               " - it is either missing or broken", vg_name)
2298
      if not test:
2299
        try:
2300
          nimg.dfree = int(nresult[constants.NV_VGLIST][vg_name])
2301
        except (ValueError, TypeError):
2302
          _ErrorIf(True, self.ENODERPC, node,
2303
                   "node returned invalid LVM info, check LVM status")
2304

    
2305
  def _CollectDiskInfo(self, nodelist, node_image, instanceinfo):
2306
    """Gets per-disk status information for all instances.
2307

2308
    @type nodelist: list of strings
2309
    @param nodelist: Node names
2310
    @type node_image: dict of (name, L{objects.Node})
2311
    @param node_image: Node objects
2312
    @type instanceinfo: dict of (name, L{objects.Instance})
2313
    @param instanceinfo: Instance objects
2314
    @rtype: {instance: {node: [(succes, payload)]}}
2315
    @return: a dictionary of per-instance dictionaries with nodes as
2316
        keys and disk information as values; the disk information is a
2317
        list of tuples (success, payload)
2318

2319
    """
2320
    _ErrorIf = self._ErrorIf # pylint: disable-msg=C0103
2321

    
2322
    node_disks = {}
2323
    node_disks_devonly = {}
2324
    diskless_instances = set()
2325
    diskless = constants.DT_DISKLESS
2326

    
2327
    for nname in nodelist:
2328
      node_instances = list(itertools.chain(node_image[nname].pinst,
2329
                                            node_image[nname].sinst))
2330
      diskless_instances.update(inst for inst in node_instances
2331
                                if instanceinfo[inst].disk_template == diskless)
2332
      disks = [(inst, disk)
2333
               for inst in node_instances
2334
               for disk in instanceinfo[inst].disks]
2335

    
2336
      if not disks:
2337
        # No need to collect data
2338
        continue
2339

    
2340
      node_disks[nname] = disks
2341

    
2342
      # Creating copies as SetDiskID below will modify the objects and that can
2343
      # lead to incorrect data returned from nodes
2344
      devonly = [dev.Copy() for (_, dev) in disks]
2345

    
2346
      for dev in devonly:
2347
        self.cfg.SetDiskID(dev, nname)
2348

    
2349
      node_disks_devonly[nname] = devonly
2350

    
2351
    assert len(node_disks) == len(node_disks_devonly)
2352

    
2353
    # Collect data from all nodes with disks
2354
    result = self.rpc.call_blockdev_getmirrorstatus_multi(node_disks.keys(),
2355
                                                          node_disks_devonly)
2356

    
2357
    assert len(result) == len(node_disks)
2358

    
2359
    instdisk = {}
2360

    
2361
    for (nname, nres) in result.items():
2362
      disks = node_disks[nname]
2363

    
2364
      if nres.offline:
2365
        # No data from this node
2366
        data = len(disks) * [(False, "node offline")]
2367
      else:
2368
        msg = nres.fail_msg
2369
        _ErrorIf(msg, self.ENODERPC, nname,
2370
                 "while getting disk information: %s", msg)
2371
        if msg:
2372
          # No data from this node
2373
          data = len(disks) * [(False, msg)]
2374
        else:
2375
          data = []
2376
          for idx, i in enumerate(nres.payload):
2377
            if isinstance(i, (tuple, list)) and len(i) == 2:
2378
              data.append(i)
2379
            else:
2380
              logging.warning("Invalid result from node %s, entry %d: %s",
2381
                              nname, idx, i)
2382
              data.append((False, "Invalid result from the remote node"))
2383

    
2384
      for ((inst, _), status) in zip(disks, data):
2385
        instdisk.setdefault(inst, {}).setdefault(nname, []).append(status)
2386

    
2387
    # Add empty entries for diskless instances.
2388
    for inst in diskless_instances:
2389
      assert inst not in instdisk
2390
      instdisk[inst] = {}
2391

    
2392
    assert compat.all(len(statuses) == len(instanceinfo[inst].disks) and
2393
                      len(nnames) <= len(instanceinfo[inst].all_nodes) and
2394
                      compat.all(isinstance(s, (tuple, list)) and
2395
                                 len(s) == 2 for s in statuses)
2396
                      for inst, nnames in instdisk.items()
2397
                      for nname, statuses in nnames.items())
2398
    assert set(instdisk) == set(instanceinfo), "instdisk consistency failure"
2399

    
2400
    return instdisk
2401

    
2402
  def BuildHooksEnv(self):
2403
    """Build hooks env.
2404

2405
    Cluster-Verify hooks just ran in the post phase and their failure makes
2406
    the output be logged in the verify output and the verification to fail.
2407

2408
    """
2409
    env = {
2410
      "CLUSTER_TAGS": " ".join(self.cfg.GetClusterInfo().GetTags())
2411
      }
2412

    
2413
    env.update(("NODE_TAGS_%s" % node.name, " ".join(node.GetTags()))
2414
               for node in self.my_node_info.values())
2415

    
2416
    return env
2417

    
2418
  def BuildHooksNodes(self):
2419
    """Build hooks nodes.
2420

2421
    """
2422
    return ([], self.my_node_names)
2423

    
2424
  def Exec(self, feedback_fn):
2425
    """Verify integrity of the node group, performing various test on nodes.
2426

2427
    """
2428
    # This method has too many local variables. pylint: disable-msg=R0914
2429

    
2430
    if not self.my_node_names:
2431
      # empty node group
2432
      feedback_fn("* Empty node group, skipping verification")
2433
      return True
2434

    
2435
    self.bad = False
2436
    _ErrorIf = self._ErrorIf # pylint: disable-msg=C0103
2437
    verbose = self.op.verbose
2438
    self._feedback_fn = feedback_fn
2439

    
2440
    vg_name = self.cfg.GetVGName()
2441
    drbd_helper = self.cfg.GetDRBDHelper()
2442
    cluster = self.cfg.GetClusterInfo()
2443
    groupinfo = self.cfg.GetAllNodeGroupsInfo()
2444
    hypervisors = cluster.enabled_hypervisors
2445
    node_data_list = [self.my_node_info[name] for name in self.my_node_names]
2446

    
2447
    i_non_redundant = [] # Non redundant instances
2448
    i_non_a_balanced = [] # Non auto-balanced instances
2449
    n_offline = 0 # Count of offline nodes
2450
    n_drained = 0 # Count of nodes being drained
2451
    node_vol_should = {}
2452

    
2453
    # FIXME: verify OS list
2454

    
2455
    # File verification
2456
    filemap = _ComputeAncillaryFiles(cluster, False)
2457

    
2458
    # do local checksums
2459
    master_node = self.master_node = self.cfg.GetMasterNode()
2460
    master_ip = self.cfg.GetMasterIP()
2461

    
2462
    feedback_fn("* Gathering data (%d nodes)" % len(self.my_node_names))
2463

    
2464
    # We will make nodes contact all nodes in their group, and one node from
2465
    # every other group.
2466
    # TODO: should it be a *random* node, different every time?
2467
    online_nodes = [node.name for node in node_data_list if not node.offline]
2468
    other_group_nodes = {}
2469

    
2470
    for name in sorted(self.all_node_info):
2471
      node = self.all_node_info[name]
2472
      if (node.group not in other_group_nodes
2473
          and node.group != self.group_uuid
2474
          and not node.offline):
2475
        other_group_nodes[node.group] = node.name
2476

    
2477
    node_verify_param = {
2478
      constants.NV_FILELIST:
2479
        utils.UniqueSequence(filename
2480
                             for files in filemap
2481
                             for filename in files),
2482
      constants.NV_NODELIST: online_nodes + other_group_nodes.values(),
2483
      constants.NV_HYPERVISOR: hypervisors,
2484
      constants.NV_HVPARAMS:
2485
        _GetAllHypervisorParameters(cluster, self.all_inst_info.values()),
2486
      constants.NV_NODENETTEST: [(node.name, node.primary_ip, node.secondary_ip)
2487
                                 for node in node_data_list
2488
                                 if not node.offline],
2489
      constants.NV_INSTANCELIST: hypervisors,
2490
      constants.NV_VERSION: None,
2491
      constants.NV_HVINFO: self.cfg.GetHypervisorType(),
2492
      constants.NV_NODESETUP: None,
2493
      constants.NV_TIME: None,
2494
      constants.NV_MASTERIP: (master_node, master_ip),
2495
      constants.NV_OSLIST: None,
2496
      constants.NV_VMNODES: self.cfg.GetNonVmCapableNodeList(),
2497
      }
2498

    
2499
    if vg_name is not None:
2500
      node_verify_param[constants.NV_VGLIST] = None
2501
      node_verify_param[constants.NV_LVLIST] = vg_name
2502
      node_verify_param[constants.NV_PVLIST] = [vg_name]
2503
      node_verify_param[constants.NV_DRBDLIST] = None
2504

    
2505
    if drbd_helper:
2506
      node_verify_param[constants.NV_DRBDHELPER] = drbd_helper
2507

    
2508
    # bridge checks
2509
    # FIXME: this needs to be changed per node-group, not cluster-wide
2510
    bridges = set()
2511
    default_nicpp = cluster.nicparams[constants.PP_DEFAULT]
2512
    if default_nicpp[constants.NIC_MODE] == constants.NIC_MODE_BRIDGED:
2513
      bridges.add(default_nicpp[constants.NIC_LINK])
2514
    for instance in self.my_inst_info.values():
2515
      for nic in instance.nics:
2516
        full_nic = cluster.SimpleFillNIC(nic.nicparams)
2517
        if full_nic[constants.NIC_MODE] == constants.NIC_MODE_BRIDGED:
2518
          bridges.add(full_nic[constants.NIC_LINK])
2519

    
2520
    if bridges:
2521
      node_verify_param[constants.NV_BRIDGES] = list(bridges)
2522

    
2523
    # Build our expected cluster state
2524
    node_image = dict((node.name, self.NodeImage(offline=node.offline,
2525
                                                 name=node.name,
2526
                                                 vm_capable=node.vm_capable))
2527
                      for node in node_data_list)
2528

    
2529
    # Gather OOB paths
2530
    oob_paths = []
2531
    for node in self.all_node_info.values():
2532
      path = _SupportsOob(self.cfg, node)
2533
      if path and path not in oob_paths:
2534
        oob_paths.append(path)
2535

    
2536
    if oob_paths:
2537
      node_verify_param[constants.NV_OOB_PATHS] = oob_paths
2538

    
2539
    for instance in self.my_inst_names:
2540
      inst_config = self.my_inst_info[instance]
2541

    
2542
      for nname in inst_config.all_nodes:
2543
        if nname not in node_image:
2544
          gnode = self.NodeImage(name=nname)
2545
          gnode.ghost = (nname not in self.all_node_info)
2546
          node_image[nname] = gnode
2547

    
2548
      inst_config.MapLVsByNode(node_vol_should)
2549

    
2550
      pnode = inst_config.primary_node
2551
      node_image[pnode].pinst.append(instance)
2552

    
2553
      for snode in inst_config.secondary_nodes:
2554
        nimg = node_image[snode]
2555
        nimg.sinst.append(instance)
2556
        if pnode not in nimg.sbp:
2557
          nimg.sbp[pnode] = []
2558
        nimg.sbp[pnode].append(instance)
2559

    
2560
    # At this point, we have the in-memory data structures complete,
2561
    # except for the runtime information, which we'll gather next
2562

    
2563
    # Due to the way our RPC system works, exact response times cannot be
2564
    # guaranteed (e.g. a broken node could run into a timeout). By keeping the
2565
    # time before and after executing the request, we can at least have a time
2566
    # window.
2567
    nvinfo_starttime = time.time()
2568
    all_nvinfo = self.rpc.call_node_verify(self.my_node_names,
2569
                                           node_verify_param,
2570
                                           self.cfg.GetClusterName())
2571
    nvinfo_endtime = time.time()
2572

    
2573
    if self.extra_lv_nodes and vg_name is not None:
2574
      extra_lv_nvinfo = \
2575
          self.rpc.call_node_verify(self.extra_lv_nodes,
2576
                                    {constants.NV_LVLIST: vg_name},
2577
                                    self.cfg.GetClusterName())
2578
    else:
2579
      extra_lv_nvinfo = {}
2580

    
2581
    all_drbd_map = self.cfg.ComputeDRBDMap()
2582

    
2583
    feedback_fn("* Gathering disk information (%s nodes)" %
2584
                len(self.my_node_names))
2585
    instdisk = self._CollectDiskInfo(self.my_node_names, node_image,
2586
                                     self.my_inst_info)
2587

    
2588
    feedback_fn("* Verifying configuration file consistency")
2589

    
2590
    # If not all nodes are being checked, we need to make sure the master node
2591
    # and a non-checked vm_capable node are in the list.
2592
    absent_nodes = set(self.all_node_info).difference(self.my_node_info)
2593
    if absent_nodes:
2594
      vf_nvinfo = all_nvinfo.copy()
2595
      vf_node_info = list(self.my_node_info.values())
2596
      additional_nodes = []
2597
      if master_node not in self.my_node_info:
2598
        additional_nodes.append(master_node)
2599
        vf_node_info.append(self.all_node_info[master_node])
2600
      # Add the first vm_capable node we find which is not included
2601
      for node in absent_nodes:
2602
        nodeinfo = self.all_node_info[node]
2603
        if nodeinfo.vm_capable and not nodeinfo.offline:
2604
          additional_nodes.append(node)
2605
          vf_node_info.append(self.all_node_info[node])
2606
          break
2607
      key = constants.NV_FILELIST
2608
      vf_nvinfo.update(self.rpc.call_node_verify(additional_nodes,
2609
                                                 {key: node_verify_param[key]},
2610
                                                 self.cfg.GetClusterName()))
2611
    else:
2612
      vf_nvinfo = all_nvinfo
2613
      vf_node_info = self.my_node_info.values()
2614

    
2615
    self._VerifyFiles(_ErrorIf, vf_node_info, master_node, vf_nvinfo, filemap)
2616

    
2617
    feedback_fn("* Verifying node status")
2618

    
2619
    refos_img = None
2620

    
2621
    for node_i in node_data_list:
2622
      node = node_i.name
2623
      nimg = node_image[node]
2624

    
2625
      if node_i.offline:
2626
        if verbose:
2627
          feedback_fn("* Skipping offline node %s" % (node,))
2628
        n_offline += 1
2629
        continue
2630

    
2631
      if node == master_node:
2632
        ntype = "master"
2633
      elif node_i.master_candidate:
2634
        ntype = "master candidate"
2635
      elif node_i.drained:
2636
        ntype = "drained"
2637
        n_drained += 1
2638
      else:
2639
        ntype = "regular"
2640
      if verbose:
2641
        feedback_fn("* Verifying node %s (%s)" % (node, ntype))
2642

    
2643
      msg = all_nvinfo[node].fail_msg
2644
      _ErrorIf(msg, self.ENODERPC, node, "while contacting node: %s", msg)
2645
      if msg:
2646
        nimg.rpc_fail = True
2647
        continue
2648

    
2649
      nresult = all_nvinfo[node].payload
2650

    
2651
      nimg.call_ok = self._VerifyNode(node_i, nresult)
2652
      self._VerifyNodeTime(node_i, nresult, nvinfo_starttime, nvinfo_endtime)
2653
      self._VerifyNodeNetwork(node_i, nresult)
2654
      self._VerifyOob(node_i, nresult)
2655

    
2656
      if nimg.vm_capable:
2657
        self._VerifyNodeLVM(node_i, nresult, vg_name)
2658
        self._VerifyNodeDrbd(node_i, nresult, self.all_inst_info, drbd_helper,
2659
                             all_drbd_map)
2660

    
2661
        self._UpdateNodeVolumes(node_i, nresult, nimg, vg_name)
2662
        self._UpdateNodeInstances(node_i, nresult, nimg)
2663
        self._UpdateNodeInfo(node_i, nresult, nimg, vg_name)
2664
        self._UpdateNodeOS(node_i, nresult, nimg)
2665

    
2666
        if not nimg.os_fail:
2667
          if refos_img is None:
2668
            refos_img = nimg
2669
          self._VerifyNodeOS(node_i, nimg, refos_img)
2670
        self._VerifyNodeBridges(node_i, nresult, bridges)
2671

    
2672
        # Check whether all running instancies are primary for the node. (This
2673
        # can no longer be done from _VerifyInstance below, since some of the
2674
        # wrong instances could be from other node groups.)
2675
        non_primary_inst = set(nimg.instances).difference(nimg.pinst)
2676

    
2677
        for inst in non_primary_inst:
2678
          test = inst in self.all_inst_info
2679
          _ErrorIf(test, self.EINSTANCEWRONGNODE, inst,
2680
                   "instance should not run on node %s", node_i.name)
2681
          _ErrorIf(not test, self.ENODEORPHANINSTANCE, node_i.name,
2682
                   "node is running unknown instance %s", inst)
2683

    
2684
    for node, result in extra_lv_nvinfo.items():
2685
      self._UpdateNodeVolumes(self.all_node_info[node], result.payload,
2686
                              node_image[node], vg_name)
2687

    
2688
    feedback_fn("* Verifying instance status")
2689
    for instance in self.my_inst_names:
2690
      if verbose:
2691
        feedback_fn("* Verifying instance %s" % instance)
2692
      inst_config = self.my_inst_info[instance]
2693
      self._VerifyInstance(instance, inst_config, node_image,
2694
                           instdisk[instance])
2695
      inst_nodes_offline = []
2696

    
2697
      pnode = inst_config.primary_node
2698
      pnode_img = node_image[pnode]
2699
      _ErrorIf(pnode_img.rpc_fail and not pnode_img.offline,
2700
               self.ENODERPC, pnode, "instance %s, connection to"
2701
               " primary node failed", instance)
2702

    
2703
      _ErrorIf(inst_config.admin_up and pnode_img.offline,
2704
               self.EINSTANCEBADNODE, instance,
2705
               "instance is marked as running and lives on offline node %s",
2706
               inst_config.primary_node)
2707

    
2708
      # If the instance is non-redundant we cannot survive losing its primary
2709
      # node, so we are not N+1 compliant. On the other hand we have no disk
2710
      # templates with more than one secondary so that situation is not well
2711
      # supported either.
2712
      # FIXME: does not support file-backed instances
2713
      if not inst_config.secondary_nodes:
2714
        i_non_redundant.append(instance)
2715

    
2716
      _ErrorIf(len(inst_config.secondary_nodes) > 1, self.EINSTANCELAYOUT,
2717
               instance, "instance has multiple secondary nodes: %s",
2718
               utils.CommaJoin(inst_config.secondary_nodes),
2719
               code=self.ETYPE_WARNING)
2720

    
2721
      if inst_config.disk_template in constants.DTS_INT_MIRROR:
2722
        pnode = inst_config.primary_node
2723
        instance_nodes = utils.NiceSort(inst_config.all_nodes)
2724
        instance_groups = {}
2725

    
2726
        for node in instance_nodes:
2727
          instance_groups.setdefault(self.all_node_info[node].group,
2728
                                     []).append(node)
2729

    
2730
        pretty_list = [
2731
          "%s (group %s)" % (utils.CommaJoin(nodes), groupinfo[group].name)
2732
          # Sort so that we always list the primary node first.
2733
          for group, nodes in sorted(instance_groups.items(),
2734
                                     key=lambda (_, nodes): pnode in nodes,
2735
                                     reverse=True)]
2736

    
2737
        self._ErrorIf(len(instance_groups) > 1, self.EINSTANCESPLITGROUPS,
2738
                      instance, "instance has primary and secondary nodes in"
2739
                      " different groups: %s", utils.CommaJoin(pretty_list),
2740
                      code=self.ETYPE_WARNING)
2741

    
2742
      if not cluster.FillBE(inst_config)[constants.BE_AUTO_BALANCE]:
2743
        i_non_a_balanced.append(instance)
2744

    
2745
      for snode in inst_config.secondary_nodes:
2746
        s_img = node_image[snode]
2747
        _ErrorIf(s_img.rpc_fail and not s_img.offline, self.ENODERPC, snode,
2748
                 "instance %s, connection to secondary node failed", instance)
2749

    
2750
        if s_img.offline:
2751
          inst_nodes_offline.append(snode)
2752

    
2753
      # warn that the instance lives on offline nodes
2754
      _ErrorIf(inst_nodes_offline, self.EINSTANCEBADNODE, instance,
2755
               "instance has offline secondary node(s) %s",
2756
               utils.CommaJoin(inst_nodes_offline))
2757
      # ... or ghost/non-vm_capable nodes
2758
      for node in inst_config.all_nodes:
2759
        _ErrorIf(node_image[node].ghost, self.EINSTANCEBADNODE, instance,
2760
                 "instance lives on ghost node %s", node)
2761
        _ErrorIf(not node_image[node].vm_capable, self.EINSTANCEBADNODE,
2762
                 instance, "instance lives on non-vm_capable node %s", node)
2763

    
2764
    feedback_fn("* Verifying orphan volumes")
2765
    reserved = utils.FieldSet(*cluster.reserved_lvs)
2766

    
2767
    # We will get spurious "unknown volume" warnings if any node of this group
2768
    # is secondary for an instance whose primary is in another group. To avoid
2769
    # them, we find these instances and add their volumes to node_vol_should.
2770
    for inst in self.all_inst_info.values():
2771
      for secondary in inst.secondary_nodes:
2772
        if (secondary in self.my_node_info
2773
            and inst.name not in self.my_inst_info):
2774
          inst.MapLVsByNode(node_vol_should)
2775
          break
2776

    
2777
    self._VerifyOrphanVolumes(node_vol_should, node_image, reserved)
2778

    
2779
    if constants.VERIFY_NPLUSONE_MEM not in self.op.skip_checks:
2780
      feedback_fn("* Verifying N+1 Memory redundancy")
2781
      self._VerifyNPlusOneMemory(node_image, self.my_inst_info)
2782

    
2783
    feedback_fn("* Other Notes")
2784
    if i_non_redundant:
2785
      feedback_fn("  - NOTICE: %d non-redundant instance(s) found."
2786
                  % len(i_non_redundant))
2787

    
2788
    if i_non_a_balanced:
2789
      feedback_fn("  - NOTICE: %d non-auto-balanced instance(s) found."
2790
                  % len(i_non_a_balanced))
2791

    
2792
    if n_offline:
2793
      feedback_fn("  - NOTICE: %d offline node(s) found." % n_offline)
2794

    
2795
    if n_drained:
2796
      feedback_fn("  - NOTICE: %d drained node(s) found." % n_drained)
2797

    
2798
    return not self.bad
2799

    
2800
  def HooksCallBack(self, phase, hooks_results, feedback_fn, lu_result):
2801
    """Analyze the post-hooks' result
2802

2803
    This method analyses the hook result, handles it, and sends some
2804
    nicely-formatted feedback back to the user.
2805

2806
    @param phase: one of L{constants.HOOKS_PHASE_POST} or
2807
        L{constants.HOOKS_PHASE_PRE}; it denotes the hooks phase
2808
    @param hooks_results: the results of the multi-node hooks rpc call
2809
    @param feedback_fn: function used send feedback back to the caller
2810
    @param lu_result: previous Exec result
2811
    @return: the new Exec result, based on the previous result
2812
        and hook results
2813

2814
    """
2815
    # We only really run POST phase hooks, only for non-empty groups,
2816
    # and are only interested in their results
2817
    if not self.my_node_names:
2818
      # empty node group
2819
      pass
2820
    elif phase == constants.HOOKS_PHASE_POST:
2821
      # Used to change hooks' output to proper indentation
2822
      feedback_fn("* Hooks Results")
2823
      assert hooks_results, "invalid result from hooks"
2824

    
2825
      for node_name in hooks_results:
2826
        res = hooks_results[node_name]
2827
        msg = res.fail_msg
2828
        test = msg and not res.offline
2829
        self._ErrorIf(test, self.ENODEHOOKS, node_name,
2830
                      "Communication failure in hooks execution: %s", msg)
2831
        if res.offline or msg:
2832
          # No need to investigate payload if node is offline or gave an error.
2833
          # override manually lu_result here as _ErrorIf only
2834
          # overrides self.bad
2835
          lu_result = 1
2836
          continue
2837
        for script, hkr, output in res.payload:
2838
          test = hkr == constants.HKR_FAIL
2839
          self._ErrorIf(test, self.ENODEHOOKS, node_name,
2840
                        "Script %s failed, output:", script)
2841
          if test:
2842
            output = self._HOOKS_INDENT_RE.sub('      ', output)
2843
            feedback_fn("%s" % output)
2844
            lu_result = 0
2845

    
2846
    return lu_result
2847

    
2848

    
2849
class LUClusterVerifyDisks(NoHooksLU):
2850
  """Verifies the cluster disks status.
2851

2852
  """
2853
  REQ_BGL = False
2854

    
2855
  def ExpandNames(self):
2856
    self.needed_locks = {
2857
      locking.LEVEL_NODE: locking.ALL_SET,
2858
      locking.LEVEL_INSTANCE: locking.ALL_SET,
2859
    }
2860
    self.share_locks = dict.fromkeys(locking.LEVELS, 1)
2861

    
2862
  def Exec(self, feedback_fn):
2863
    """Verify integrity of cluster disks.
2864

2865
    @rtype: tuple of three items
2866
    @return: a tuple of (dict of node-to-node_error, list of instances
2867
        which need activate-disks, dict of instance: (node, volume) for
2868
        missing volumes
2869

2870
    """
2871
    result = res_nodes, res_instances, res_missing = {}, [], {}
2872

    
2873
    nodes = utils.NiceSort(self.cfg.GetVmCapableNodeList())
2874
    instances = self.cfg.GetAllInstancesInfo().values()
2875

    
2876
    nv_dict = {}
2877
    for inst in instances:
2878
      inst_lvs = {}
2879
      if not inst.admin_up:
2880
        continue
2881
      inst.MapLVsByNode(inst_lvs)
2882
      # transform { iname: {node: [vol,],},} to {(node, vol): iname}
2883
      for node, vol_list in inst_lvs.iteritems():
2884
        for vol in vol_list:
2885
          nv_dict[(node, vol)] = inst
2886

    
2887
    if not nv_dict:
2888
      return result
2889

    
2890
    node_lvs = self.rpc.call_lv_list(nodes, [])
2891
    for node, node_res in node_lvs.items():
2892
      if node_res.offline:
2893
        continue
2894
      msg = node_res.fail_msg
2895
      if msg:
2896
        logging.warning("Error enumerating LVs on node %s: %s", node, msg)
2897
        res_nodes[node] = msg
2898
        continue
2899

    
2900
      lvs = node_res.payload
2901
      for lv_name, (_, _, lv_online) in lvs.items():
2902
        inst = nv_dict.pop((node, lv_name), None)
2903
        if (not lv_online and inst is not None
2904
            and inst.name not in res_instances):
2905
          res_instances.append(inst.name)
2906

    
2907
    # any leftover items in nv_dict are missing LVs, let's arrange the
2908
    # data better
2909
    for key, inst in nv_dict.iteritems():
2910
      if inst.name not in res_missing:
2911
        res_missing[inst.name] = []
2912
      res_missing[inst.name].append(key)
2913

    
2914
    return result
2915

    
2916

    
2917
class LUClusterRepairDiskSizes(NoHooksLU):
2918
  """Verifies the cluster disks sizes.
2919

2920
  """
2921
  REQ_BGL = False
2922

    
2923
  def ExpandNames(self):
2924
    if self.op.instances:
2925
      self.wanted_names = _GetWantedInstances(self, self.op.instances)
2926
      self.needed_locks = {
2927
        locking.LEVEL_NODE: [],
2928
        locking.LEVEL_INSTANCE: self.wanted_names,
2929
        }
2930
      self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
2931
    else:
2932
      self.wanted_names = None
2933
      self.needed_locks = {
2934
        locking.LEVEL_NODE: locking.ALL_SET,
2935
        locking.LEVEL_INSTANCE: locking.ALL_SET,
2936
        }
2937
    self.share_locks = dict.fromkeys(locking.LEVELS, 1)
2938

    
2939
  def DeclareLocks(self, level):
2940
    if level == locking.LEVEL_NODE and self.wanted_names is not None:
2941
      self._LockInstancesNodes(primary_only=True)
2942

    
2943
  def CheckPrereq(self):
2944
    """Check prerequisites.
2945

2946
    This only checks the optional instance list against the existing names.
2947

2948
    """
2949
    if self.wanted_names is None:
2950
      self.wanted_names = self.glm.list_owned(locking.LEVEL_INSTANCE)
2951

    
2952
    self.wanted_instances = [self.cfg.GetInstanceInfo(name) for name
2953
                             in self.wanted_names]
2954

    
2955
  def _EnsureChildSizes(self, disk):
2956
    """Ensure children of the disk have the needed disk size.
2957

2958
    This is valid mainly for DRBD8 and fixes an issue where the
2959
    children have smaller disk size.
2960

2961
    @param disk: an L{ganeti.objects.Disk} object
2962

2963
    """
2964
    if disk.dev_type == constants.LD_DRBD8:
2965
      assert disk.children, "Empty children for DRBD8?"
2966
      fchild = disk.children[0]
2967
      mismatch = fchild.size < disk.size
2968
      if mismatch:
2969
        self.LogInfo("Child disk has size %d, parent %d, fixing",
2970
                     fchild.size, disk.size)
2971
        fchild.size = disk.size
2972

    
2973
      # and we recurse on this child only, not on the metadev
2974
      return self._EnsureChildSizes(fchild) or mismatch
2975
    else:
2976
      return False
2977

    
2978
  def Exec(self, feedback_fn):
2979
    """Verify the size of cluster disks.
2980

2981
    """
2982
    # TODO: check child disks too
2983
    # TODO: check differences in size between primary/secondary nodes
2984
    per_node_disks = {}
2985
    for instance in self.wanted_instances:
2986
      pnode = instance.primary_node
2987
      if pnode not in per_node_disks:
2988
        per_node_disks[pnode] = []
2989
      for idx, disk in enumerate(instance.disks):
2990
        per_node_disks[pnode].append((instance, idx, disk))
2991

    
2992
    changed = []
2993
    for node, dskl in per_node_disks.items():
2994
      newl = [v[2].Copy() for v in dskl]
2995
      for dsk in newl:
2996
        self.cfg.SetDiskID(dsk, node)
2997
      result = self.rpc.call_blockdev_getsize(node, newl)
2998
      if result.fail_msg:
2999
        self.LogWarning("Failure in blockdev_getsize call to node"
3000
                        " %s, ignoring", node)
3001
        continue
3002
      if len(result.payload) != len(dskl):
3003
        logging.warning("Invalid result from node %s: len(dksl)=%d,"
3004
                        " result.payload=%s", node, len(dskl), result.payload)
3005
        self.LogWarning("Invalid result from node %s, ignoring node results",
3006
                        node)
3007
        continue
3008
      for ((instance, idx, disk), size) in zip(dskl, result.payload):
3009
        if size is None:
3010
          self.LogWarning("Disk %d of instance %s did not return size"
3011
                          " information, ignoring", idx, instance.name)
3012
          continue
3013
        if not isinstance(size, (int, long)):
3014
          self.LogWarning("Disk %d of instance %s did not return valid"
3015
                          " size information, ignoring", idx, instance.name)
3016
          continue
3017
        size = size >> 20
3018
        if size != disk.size:
3019
          self.LogInfo("Disk %d of instance %s has mismatched size,"
3020
                       " correcting: recorded %d, actual %d", idx,
3021
                       instance.name, disk.size, size)
3022
          disk.size = size
3023
          self.cfg.Update(instance, feedback_fn)
3024
          changed.append((instance.name, idx, size))
3025
        if self._EnsureChildSizes(disk):
3026
          self.cfg.Update(instance, feedback_fn)
3027
          changed.append((instance.name, idx, disk.size))
3028
    return changed
3029

    
3030

    
3031
class LUClusterRename(LogicalUnit):
3032
  """Rename the cluster.
3033

3034
  """
3035
  HPATH = "cluster-rename"
3036
  HTYPE = constants.HTYPE_CLUSTER
3037

    
3038
  def BuildHooksEnv(self):
3039
    """Build hooks env.
3040

3041
    """
3042
    return {
3043
      "OP_TARGET": self.cfg.GetClusterName(),
3044
      "NEW_NAME": self.op.name,
3045
      }
3046

    
3047
  def BuildHooksNodes(self):
3048
    """Build hooks nodes.
3049

3050
    """
3051
    return ([self.cfg.GetMasterNode()], self.cfg.GetNodeList())
3052

    
3053
  def CheckPrereq(self):
3054
    """Verify that the passed name is a valid one.
3055

3056
    """
3057
    hostname = netutils.GetHostname(name=self.op.name,
3058
                                    family=self.cfg.GetPrimaryIPFamily())
3059

    
3060
    new_name = hostname.name
3061
    self.ip = new_ip = hostname.ip
3062
    old_name = self.cfg.GetClusterName()
3063
    old_ip = self.cfg.GetMasterIP()
3064
    if new_name == old_name and new_ip == old_ip:
3065
      raise errors.OpPrereqError("Neither the name nor the IP address of the"
3066
                                 " cluster has changed",
3067
                                 errors.ECODE_INVAL)
3068
    if new_ip != old_ip:
3069
      if netutils.TcpPing(new_ip, constants.DEFAULT_NODED_PORT):
3070
        raise errors.OpPrereqError("The given cluster IP address (%s) is"
3071
                                   " reachable on the network" %
3072
                                   new_ip, errors.ECODE_NOTUNIQUE)
3073

    
3074
    self.op.name = new_name
3075

    
3076
  def Exec(self, feedback_fn):
3077
    """Rename the cluster.
3078

3079
    """
3080
    clustername = self.op.name
3081
    ip = self.ip
3082

    
3083
    # shutdown the master IP
3084
    master = self.cfg.GetMasterNode()
3085
    result = self.rpc.call_node_stop_master(master, False)
3086
    result.Raise("Could not disable the master role")
3087

    
3088
    try:
3089
      cluster = self.cfg.GetClusterInfo()
3090
      cluster.cluster_name = clustername
3091
      cluster.master_ip = ip
3092
      self.cfg.Update(cluster, feedback_fn)
3093

    
3094
      # update the known hosts file
3095
      ssh.WriteKnownHostsFile(self.cfg, constants.SSH_KNOWN_HOSTS_FILE)
3096
      node_list = self.cfg.GetOnlineNodeList()
3097
      try:
3098
        node_list.remove(master)
3099
      except ValueError:
3100
        pass
3101
      _UploadHelper(self, node_list, constants.SSH_KNOWN_HOSTS_FILE)
3102
    finally:
3103
      result = self.rpc.call_node_start_master(master, False, False)
3104
      msg = result.fail_msg
3105
      if msg:
3106
        self.LogWarning("Could not re-enable the master role on"
3107
                        " the master, please restart manually: %s", msg)
3108

    
3109
    return clustername
3110

    
3111

    
3112
class LUClusterSetParams(LogicalUnit):
3113
  """Change the parameters of the cluster.
3114

3115
  """
3116
  HPATH = "cluster-modify"
3117
  HTYPE = constants.HTYPE_CLUSTER
3118
  REQ_BGL = False
3119

    
3120
  def CheckArguments(self):
3121
    """Check parameters
3122

3123
    """
3124
    if self.op.uid_pool:
3125
      uidpool.CheckUidPool(self.op.uid_pool)
3126

    
3127
    if self.op.add_uids:
3128
      uidpool.CheckUidPool(self.op.add_uids)
3129

    
3130
    if self.op.remove_uids:
3131
      uidpool.CheckUidPool(self.op.remove_uids)
3132

    
3133
  def ExpandNames(self):
3134
    # FIXME: in the future maybe other cluster params won't require checking on
3135
    # all nodes to be modified.
3136
    self.needed_locks = {
3137
      locking.LEVEL_NODE: locking.ALL_SET,
3138
    }
3139
    self.share_locks[locking.LEVEL_NODE] = 1
3140

    
3141
  def BuildHooksEnv(self):
3142
    """Build hooks env.
3143

3144
    """
3145
    return {
3146
      "OP_TARGET": self.cfg.GetClusterName(),
3147
      "NEW_VG_NAME": self.op.vg_name,
3148
      }
3149

    
3150
  def BuildHooksNodes(self):
3151
    """Build hooks nodes.
3152

3153
    """
3154
    mn = self.cfg.GetMasterNode()
3155
    return ([mn], [mn])
3156

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

3160
    This checks whether the given params don't conflict and
3161
    if the given volume group is valid.
3162

3163
    """
3164
    if self.op.vg_name is not None and not self.op.vg_name:
3165
      if self.cfg.HasAnyDiskOfType(constants.LD_LV):
3166
        raise errors.OpPrereqError("Cannot disable lvm storage while lvm-based"
3167
                                   " instances exist", errors.ECODE_INVAL)
3168

    
3169
    if self.op.drbd_helper is not None and not self.op.drbd_helper:
3170
      if self.cfg.HasAnyDiskOfType(constants.LD_DRBD8):
3171
        raise errors.OpPrereqError("Cannot disable drbd helper while"
3172
                                   " drbd-based instances exist",
3173
                                   errors.ECODE_INVAL)
3174

    
3175
    node_list = self.glm.list_owned(locking.LEVEL_NODE)
3176

    
3177
    # if vg_name not None, checks given volume group on all nodes
3178
    if self.op.vg_name:
3179
      vglist = self.rpc.call_vg_list(node_list)
3180
      for node in node_list:
3181
        msg = vglist[node].fail_msg
3182
        if msg:
3183
          # ignoring down node
3184
          self.LogWarning("Error while gathering data on node %s"
3185
                          " (ignoring node): %s", node, msg)
3186
          continue
3187
        vgstatus = utils.CheckVolumeGroupSize(vglist[node].payload,
3188
                                              self.op.vg_name,
3189
                                              constants.MIN_VG_SIZE)
3190
        if vgstatus:
3191
          raise errors.OpPrereqError("Error on node '%s': %s" %
3192
                                     (node, vgstatus), errors.ECODE_ENVIRON)
3193

    
3194
    if self.op.drbd_helper:
3195
      # checks given drbd helper on all nodes
3196
      helpers = self.rpc.call_drbd_helper(node_list)
3197
      for node in node_list:
3198
        ninfo = self.cfg.GetNodeInfo(node)
3199
        if ninfo.offline:
3200
          self.LogInfo("Not checking drbd helper on offline node %s", node)
3201
          continue
3202
        msg = helpers[node].fail_msg
3203
        if msg:
3204
          raise errors.OpPrereqError("Error checking drbd helper on node"
3205
                                     " '%s': %s" % (node, msg),
3206
                                     errors.ECODE_ENVIRON)
3207
        node_helper = helpers[node].payload
3208
        if node_helper != self.op.drbd_helper:
3209
          raise errors.OpPrereqError("Error on node '%s': drbd helper is %s" %
3210
                                     (node, node_helper), errors.ECODE_ENVIRON)
3211

    
3212
    self.cluster = cluster = self.cfg.GetClusterInfo()
3213
    # validate params changes
3214
    if self.op.beparams:
3215
      utils.ForceDictType(self.op.beparams, constants.BES_PARAMETER_TYPES)
3216
      self.new_beparams = cluster.SimpleFillBE(self.op.beparams)
3217

    
3218
    if self.op.ndparams:
3219
      utils.ForceDictType(self.op.ndparams, constants.NDS_PARAMETER_TYPES)
3220
      self.new_ndparams = cluster.SimpleFillND(self.op.ndparams)
3221

    
3222
      # TODO: we need a more general way to handle resetting
3223
      # cluster-level parameters to default values
3224
      if self.new_ndparams["oob_program"] == "":
3225
        self.new_ndparams["oob_program"] = \
3226
            constants.NDC_DEFAULTS[constants.ND_OOB_PROGRAM]
3227

    
3228
    if self.op.nicparams:
3229
      utils.ForceDictType(self.op.nicparams, constants.NICS_PARAMETER_TYPES)
3230
      self.new_nicparams = cluster.SimpleFillNIC(self.op.nicparams)
3231
      objects.NIC.CheckParameterSyntax(self.new_nicparams)
3232
      nic_errors = []
3233

    
3234
      # check all instances for consistency
3235
      for instance in self.cfg.GetAllInstancesInfo().values():
3236
        for nic_idx, nic in enumerate(instance.nics):
3237
          params_copy = copy.deepcopy(nic.nicparams)
3238
          params_filled = objects.FillDict(self.new_nicparams, params_copy)
3239

    
3240
          # check parameter syntax
3241
          try:
3242
            objects.NIC.CheckParameterSyntax(params_filled)
3243
          except errors.ConfigurationError, err:
3244
            nic_errors.append("Instance %s, nic/%d: %s" %
3245
                              (instance.name, nic_idx, err))
3246

    
3247
          # if we're moving instances to routed, check that they have an ip
3248
          target_mode = params_filled[constants.NIC_MODE]
3249
          if target_mode == constants.NIC_MODE_ROUTED and not nic.ip:
3250
            nic_errors.append("Instance %s, nic/%d: routed NIC with no ip"
3251
                              " address" % (instance.name, nic_idx))
3252
      if nic_errors:
3253
        raise errors.OpPrereqError("Cannot apply the change, errors:\n%s" %
3254
                                   "\n".join(nic_errors))
3255

    
3256
    # hypervisor list/parameters
3257
    self.new_hvparams = new_hvp = objects.FillDict(cluster.hvparams, {})
3258
    if self.op.hvparams:
3259
      for hv_name, hv_dict in self.op.hvparams.items():
3260
        if hv_name not in self.new_hvparams:
3261
          self.new_hvparams[hv_name] = hv_dict
3262
        else:
3263
          self.new_hvparams[hv_name].update(hv_dict)
3264

    
3265
    # os hypervisor parameters
3266
    self.new_os_hvp = objects.FillDict(cluster.os_hvp, {})
3267
    if self.op.os_hvp:
3268
      for os_name, hvs in self.op.os_hvp.items():
3269
        if os_name not in self.new_os_hvp:
3270
          self.new_os_hvp[os_name] = hvs
3271
        else:
3272
          for hv_name, hv_dict in hvs.items():
3273
            if hv_name not in self.new_os_hvp[os_name]:
3274
              self.new_os_hvp[os_name][hv_name] = hv_dict
3275
            else:
3276
              self.new_os_hvp[os_name][hv_name].update(hv_dict)
3277

    
3278
    # os parameters
3279
    self.new_osp = objects.FillDict(cluster.osparams, {})
3280
    if self.op.osparams:
3281
      for os_name, osp in self.op.osparams.items():
3282
        if os_name not in self.new_osp:
3283
          self.new_osp[os_name] = {}
3284

    
3285
        self.new_osp[os_name] = _GetUpdatedParams(self.new_osp[os_name], osp,
3286
                                                  use_none=True)
3287

    
3288
        if not self.new_osp[os_name]:
3289
          # we removed all parameters
3290
          del self.new_osp[os_name]
3291
        else:
3292
          # check the parameter validity (remote check)
3293
          _CheckOSParams(self, False, [self.cfg.GetMasterNode()],
3294
                         os_name, self.new_osp[os_name])
3295

    
3296
    # changes to the hypervisor list
3297
    if self.op.enabled_hypervisors is not None:
3298
      self.hv_list = self.op.enabled_hypervisors
3299
      for hv in self.hv_list:
3300
        # if the hypervisor doesn't already exist in the cluster
3301
        # hvparams, we initialize it to empty, and then (in both
3302
        # cases) we make sure to fill the defaults, as we might not
3303
        # have a complete defaults list if the hypervisor wasn't
3304
        # enabled before
3305
        if hv not in new_hvp:
3306
          new_hvp[hv] = {}
3307
        new_hvp[hv] = objects.FillDict(constants.HVC_DEFAULTS[hv], new_hvp[hv])
3308
        utils.ForceDictType(new_hvp[hv], constants.HVS_PARAMETER_TYPES)
3309
    else:
3310
      self.hv_list = cluster.enabled_hypervisors
3311

    
3312
    if self.op.hvparams or self.op.enabled_hypervisors is not None:
3313
      # either the enabled list has changed, or the parameters have, validate
3314
      for hv_name, hv_params in self.new_hvparams.items():
3315
        if ((self.op.hvparams and hv_name in self.op.hvparams) or
3316
            (self.op.enabled_hypervisors and
3317
             hv_name in self.op.enabled_hypervisors)):
3318
          # either this is a new hypervisor, or its parameters have changed
3319
          hv_class = hypervisor.GetHypervisor(hv_name)
3320
          utils.ForceDictType(hv_params, constants.HVS_PARAMETER_TYPES)
3321
          hv_class.CheckParameterSyntax(hv_params)
3322
          _CheckHVParams(self, node_list, hv_name, hv_params)
3323

    
3324
    if self.op.os_hvp:
3325
      # no need to check any newly-enabled hypervisors, since the
3326
      # defaults have already been checked in the above code-block
3327
      for os_name, os_hvp in self.new_os_hvp.items():
3328
        for hv_name, hv_params in os_hvp.items():
3329
          utils.ForceDictType(hv_params, constants.HVS_PARAMETER_TYPES)
3330
          # we need to fill in the new os_hvp on top of the actual hv_p
3331
          cluster_defaults = self.new_hvparams.get(hv_name, {})
3332
          new_osp = objects.FillDict(cluster_defaults, hv_params)
3333
          hv_class = hypervisor.GetHypervisor(hv_name)
3334
          hv_class.CheckParameterSyntax(new_osp)
3335
          _CheckHVParams(self, node_list, hv_name, new_osp)
3336

    
3337
    if self.op.default_iallocator:
3338
      alloc_script = utils.FindFile(self.op.default_iallocator,
3339
                                    constants.IALLOCATOR_SEARCH_PATH,
3340
                                    os.path.isfile)
3341
      if alloc_script is None:
3342
        raise errors.OpPrereqError("Invalid default iallocator script '%s'"
3343
                                   " specified" % self.op.default_iallocator,
3344
                                   errors.ECODE_INVAL)
3345

    
3346
  def Exec(self, feedback_fn):
3347
    """Change the parameters of the cluster.
3348

3349
    """
3350
    if self.op.vg_name is not None:
3351
      new_volume = self.op.vg_name
3352
      if not new_volume:
3353
        new_volume = None
3354
      if new_volume != self.cfg.GetVGName():
3355
        self.cfg.SetVGName(new_volume)
3356
      else:
3357
        feedback_fn("Cluster LVM configuration already in desired"
3358
                    " state, not changing")
3359
    if self.op.drbd_helper is not None:
3360
      new_helper = self.op.drbd_helper
3361
      if not new_helper:
3362
        new_helper = None
3363
      if new_helper != self.cfg.GetDRBDHelper():
3364
        self.cfg.SetDRBDHelper(new_helper)
3365
      else:
3366
        feedback_fn("Cluster DRBD helper already in desired state,"
3367
                    " not changing")
3368
    if self.op.hvparams:
3369
      self.cluster.hvparams = self.new_hvparams
3370
    if self.op.os_hvp:
3371
      self.cluster.os_hvp = self.new_os_hvp
3372
    if self.op.enabled_hypervisors is not None:
3373
      self.cluster.hvparams = self.new_hvparams
3374
      self.cluster.enabled_hypervisors = self.op.enabled_hypervisors
3375
    if self.op.beparams:
3376
      self.cluster.beparams[constants.PP_DEFAULT] = self.new_beparams
3377
    if self.op.nicparams:
3378
      self.cluster.nicparams[constants.PP_DEFAULT] = self.new_nicparams
3379
    if self.op.osparams:
3380
      self.cluster.osparams = self.new_osp
3381
    if self.op.ndparams:
3382
      self.cluster.ndparams = self.new_ndparams
3383

    
3384
    if self.op.candidate_pool_size is not None:
3385
      self.cluster.candidate_pool_size = self.op.candidate_pool_size
3386
      # we need to update the pool size here, otherwise the save will fail
3387
      _AdjustCandidatePool(self, [])
3388

    
3389
    if self.op.maintain_node_health is not None:
3390
      self.cluster.maintain_node_health = self.op.maintain_node_health
3391

    
3392
    if self.op.prealloc_wipe_disks is not None:
3393
      self.cluster.prealloc_wipe_disks = self.op.prealloc_wipe_disks
3394

    
3395
    if self.op.add_uids is not None:
3396
      uidpool.AddToUidPool(self.cluster.uid_pool, self.op.add_uids)
3397

    
3398
    if self.op.remove_uids is not None:
3399
      uidpool.RemoveFromUidPool(self.cluster.uid_pool, self.op.remove_uids)
3400

    
3401
    if self.op.uid_pool is not None:
3402
      self.cluster.uid_pool = self.op.uid_pool
3403

    
3404
    if self.op.default_iallocator is not None:
3405
      self.cluster.default_iallocator = self.op.default_iallocator
3406

    
3407
    if self.op.reserved_lvs is not None:
3408
      self.cluster.reserved_lvs = self.op.reserved_lvs
3409

    
3410
    def helper_os(aname, mods, desc):
3411
      desc += " OS list"
3412
      lst = getattr(self.cluster, aname)
3413
      for key, val in mods:
3414
        if key == constants.DDM_ADD:
3415
          if val in lst:
3416
            feedback_fn("OS %s already in %s, ignoring" % (val, desc))
3417
          else:
3418
            lst.append(val)
3419
        elif key == constants.DDM_REMOVE:
3420
          if val in lst:
3421
            lst.remove(val)
3422
          else:
3423
            feedback_fn("OS %s not found in %s, ignoring" % (val, desc))
3424
        else:
3425
          raise errors.ProgrammerError("Invalid modification '%s'" % key)
3426

    
3427
    if self.op.hidden_os:
3428
      helper_os("hidden_os", self.op.hidden_os, "hidden")
3429

    
3430
    if self.op.blacklisted_os:
3431
      helper_os("blacklisted_os", self.op.blacklisted_os, "blacklisted")
3432

    
3433
    if self.op.master_netdev:
3434
      master = self.cfg.GetMasterNode()
3435
      feedback_fn("Shutting down master ip on the current netdev (%s)" %
3436
                  self.cluster.master_netdev)
3437
      result = self.rpc.call_node_stop_master(master, False)
3438
      result.Raise("Could not disable the master ip")
3439
      feedback_fn("Changing master_netdev from %s to %s" %
3440
                  (self.cluster.master_netdev, self.op.master_netdev))
3441
      self.cluster.master_netdev = self.op.master_netdev
3442

    
3443
    self.cfg.Update(self.cluster, feedback_fn)
3444

    
3445
    if self.op.master_netdev:
3446
      feedback_fn("Starting the master ip on the new master netdev (%s)" %
3447
                  self.op.master_netdev)
3448
      result = self.rpc.call_node_start_master(master, False, False)
3449
      if result.fail_msg:
3450
        self.LogWarning("Could not re-enable the master ip on"
3451
                        " the master, please restart manually: %s",
3452
                        result.fail_msg)
3453

    
3454

    
3455
def _UploadHelper(lu, nodes, fname):
3456
  """Helper for uploading a file and showing warnings.
3457

3458
  """
3459
  if os.path.exists(fname):
3460
    result = lu.rpc.call_upload_file(nodes, fname)
3461
    for to_node, to_result in result.items():
3462
      msg = to_result.fail_msg
3463
      if msg:
3464
        msg = ("Copy of file %s to node %s failed: %s" %
3465
               (fname, to_node, msg))
3466
        lu.proc.LogWarning(msg)
3467

    
3468

    
3469
def _ComputeAncillaryFiles(cluster, redist):
3470
  """Compute files external to Ganeti which need to be consistent.
3471

3472
  @type redist: boolean
3473
  @param redist: Whether to include files which need to be redistributed
3474

3475
  """
3476
  # Compute files for all nodes
3477
  files_all = set([
3478
    constants.SSH_KNOWN_HOSTS_FILE,
3479
    constants.CONFD_HMAC_KEY,
3480
    constants.CLUSTER_DOMAIN_SECRET_FILE,
3481
    ])
3482

    
3483
  if not redist:
3484
    files_all.update(constants.ALL_CERT_FILES)
3485
    files_all.update(ssconf.SimpleStore().GetFileList())
3486

    
3487
  if cluster.modify_etc_hosts:
3488
    files_all.add(constants.ETC_HOSTS)
3489

    
3490
  # Files which must either exist on all nodes or on none
3491
  files_all_opt = set([
3492
    constants.RAPI_USERS_FILE,
3493
    ])
3494

    
3495
  # Files which should only be on master candidates
3496
  files_mc = set()
3497
  if not redist:
3498
    files_mc.add(constants.CLUSTER_CONF_FILE)
3499

    
3500
  # Files which should only be on VM-capable nodes
3501
  files_vm = set(filename
3502
    for hv_name in cluster.enabled_hypervisors
3503
    for filename in hypervisor.GetHypervisor(hv_name).GetAncillaryFiles())
3504

    
3505
  # Filenames must be unique
3506
  assert (len(files_all | files_all_opt | files_mc | files_vm) ==
3507
          sum(map(len, [files_all, files_all_opt, files_mc, files_vm]))), \
3508
         "Found file listed in more than one file list"
3509

    
3510
  return (files_all, files_all_opt, files_mc, files_vm)
3511

    
3512

    
3513
def _RedistributeAncillaryFiles(lu, additional_nodes=None, additional_vm=True):
3514
  """Distribute additional files which are part of the cluster configuration.
3515

3516
  ConfigWriter takes care of distributing the config and ssconf files, but
3517
  there are more files which should be distributed to all nodes. This function
3518
  makes sure those are copied.
3519

3520
  @param lu: calling logical unit
3521
  @param additional_nodes: list of nodes not in the config to distribute to
3522
  @type additional_vm: boolean
3523
  @param additional_vm: whether the additional nodes are vm-capable or not
3524

3525
  """
3526
  # Gather target nodes
3527
  cluster = lu.cfg.GetClusterInfo()
3528
  master_info = lu.cfg.GetNodeInfo(lu.cfg.GetMasterNode())
3529

    
3530
  online_nodes = lu.cfg.GetOnlineNodeList()
3531
  vm_nodes = lu.cfg.GetVmCapableNodeList()
3532

    
3533
  if additional_nodes is not None:
3534
    online_nodes.extend(additional_nodes)
3535
    if additional_vm:
3536
      vm_nodes.extend(additional_nodes)
3537

    
3538
  # Never distribute to master node
3539
  for nodelist in [online_nodes, vm_nodes]:
3540
    if master_info.name in nodelist:
3541
      nodelist.remove(master_info.name)
3542

    
3543
  # Gather file lists
3544
  (files_all, files_all_opt, files_mc, files_vm) = \
3545
    _ComputeAncillaryFiles(cluster, True)
3546

    
3547
  # Never re-distribute configuration file from here
3548
  assert not (constants.CLUSTER_CONF_FILE in files_all or
3549
              constants.CLUSTER_CONF_FILE in files_vm)
3550
  assert not files_mc, "Master candidates not handled in this function"
3551

    
3552
  filemap = [
3553
    (online_nodes, files_all),
3554
    (online_nodes, files_all_opt),
3555
    (vm_nodes, files_vm),
3556
    ]
3557

    
3558
  # Upload the files
3559
  for (node_list, files) in filemap:
3560
    for fname in files:
3561
      _UploadHelper(lu, node_list, fname)
3562

    
3563

    
3564
class LUClusterRedistConf(NoHooksLU):
3565
  """Force the redistribution of cluster configuration.
3566

3567
  This is a very simple LU.
3568

3569
  """
3570
  REQ_BGL = False
3571

    
3572
  def ExpandNames(self):
3573
    self.needed_locks = {
3574
      locking.LEVEL_NODE: locking.ALL_SET,
3575
    }
3576
    self.share_locks[locking.LEVEL_NODE] = 1
3577

    
3578
  def Exec(self, feedback_fn):
3579
    """Redistribute the configuration.
3580

3581
    """
3582
    self.cfg.Update(self.cfg.GetClusterInfo(), feedback_fn)
3583
    _RedistributeAncillaryFiles(self)
3584

    
3585

    
3586
def _WaitForSync(lu, instance, disks=None, oneshot=False):
3587
  """Sleep and poll for an instance's disk to sync.
3588

3589
  """
3590
  if not instance.disks or disks is not None and not disks:
3591
    return True
3592

    
3593
  disks = _ExpandCheckDisks(instance, disks)
3594

    
3595
  if not oneshot:
3596
    lu.proc.LogInfo("Waiting for instance %s to sync disks." % instance.name)
3597

    
3598
  node = instance.primary_node
3599

    
3600
  for dev in disks:
3601
    lu.cfg.SetDiskID(dev, node)
3602

    
3603
  # TODO: Convert to utils.Retry
3604

    
3605
  retries = 0
3606
  degr_retries = 10 # in seconds, as we sleep 1 second each time
3607
  while True:
3608
    max_time = 0
3609
    done = True
3610
    cumul_degraded = False
3611
    rstats = lu.rpc.call_blockdev_getmirrorstatus(node, disks)
3612
    msg = rstats.fail_msg
3613
    if msg:
3614
      lu.LogWarning("Can't get any data from node %s: %s", node, msg)
3615
      retries += 1
3616
      if retries >= 10:
3617
        raise errors.RemoteError("Can't contact node %s for mirror data,"
3618
                                 " aborting." % node)
3619
      time.sleep(6)
3620
      continue
3621
    rstats = rstats.payload
3622
    retries = 0
3623
    for i, mstat in enumerate(rstats):
3624
      if mstat is None:
3625
        lu.LogWarning("Can't compute data for node %s/%s",
3626
                           node, disks[i].iv_name)
3627
        continue
3628

    
3629
      cumul_degraded = (cumul_degraded or
3630
                        (mstat.is_degraded and mstat.sync_percent is None))
3631
      if mstat.sync_percent is not None:
3632
        done = False
3633
        if mstat.estimated_time is not None:
3634
          rem_time = ("%s remaining (estimated)" %
3635
                      utils.FormatSeconds(mstat.estimated_time))
3636
          max_time = mstat.estimated_time
3637
        else:
3638
          rem_time = "no time estimate"
3639
        lu.proc.LogInfo("- device %s: %5.2f%% done, %s" %
3640
                        (disks[i].iv_name, mstat.sync_percent, rem_time))
3641

    
3642
    # if we're done but degraded, let's do a few small retries, to
3643
    # make sure we see a stable and not transient situation; therefore
3644
    # we force restart of the loop
3645
    if (done or oneshot) and cumul_degraded and degr_retries > 0:
3646
      logging.info("Degraded disks found, %d retries left", degr_retries)
3647
      degr_retries -= 1
3648
      time.sleep(1)
3649
      continue
3650

    
3651
    if done or oneshot:
3652
      break
3653

    
3654
    time.sleep(min(60, max_time))
3655

    
3656
  if done:
3657
    lu.proc.LogInfo("Instance %s's disks are in sync." % instance.name)
3658
  return not cumul_degraded
3659

    
3660

    
3661
def _CheckDiskConsistency(lu, dev, node, on_primary, ldisk=False):
3662
  """Check that mirrors are not degraded.
3663

3664
  The ldisk parameter, if True, will change the test from the
3665
  is_degraded attribute (which represents overall non-ok status for
3666
  the device(s)) to the ldisk (representing the local storage status).
3667

3668
  """
3669
  lu.cfg.SetDiskID(dev, node)
3670

    
3671
  result = True
3672

    
3673
  if on_primary or dev.AssembleOnSecondary():
3674
    rstats = lu.rpc.call_blockdev_find(node, dev)
3675
    msg = rstats.fail_msg
3676
    if msg:
3677
      lu.LogWarning("Can't find disk on node %s: %s", node, msg)
3678
      result = False
3679
    elif not rstats.payload:
3680
      lu.LogWarning("Can't find disk on node %s", node)
3681
      result = False
3682
    else:
3683
      if ldisk:
3684
        result = result and rstats.payload.ldisk_status == constants.LDS_OKAY
3685
      else:
3686
        result = result and not rstats.payload.is_degraded
3687

    
3688
  if dev.children:
3689
    for child in dev.children:
3690
      result = result and _CheckDiskConsistency(lu, child, node, on_primary)
3691

    
3692
  return result
3693

    
3694

    
3695
class LUOobCommand(NoHooksLU):
3696
  """Logical unit for OOB handling.
3697

3698
  """
3699
  REG_BGL = False
3700
  _SKIP_MASTER = (constants.OOB_POWER_OFF, constants.OOB_POWER_CYCLE)
3701

    
3702
  def ExpandNames(self):
3703
    """Gather locks we need.
3704

3705
    """
3706
    if self.op.node_names:
3707
      self.op.node_names = _GetWantedNodes(self, self.op.node_names)
3708
      lock_names = self.op.node_names
3709
    else:
3710
      lock_names = locking.ALL_SET
3711

    
3712
    self.needed_locks = {
3713
      locking.LEVEL_NODE: lock_names,
3714
      }
3715

    
3716
  def CheckPrereq(self):
3717
    """Check prerequisites.
3718

3719
    This checks:
3720
     - the node exists in the configuration
3721
     - OOB is supported
3722

3723
    Any errors are signaled by raising errors.OpPrereqError.
3724

3725
    """
3726
    self.nodes = []
3727
    self.master_node = self.cfg.GetMasterNode()
3728

    
3729
    assert self.op.power_delay >= 0.0
3730

    
3731
    if self.op.node_names:
3732
      if (self.op.command in self._SKIP_MASTER and
3733
          self.master_node in self.op.node_names):
3734
        master_node_obj = self.cfg.GetNodeInfo(self.master_node)
3735
        master_oob_handler = _SupportsOob(self.cfg, master_node_obj)
3736

    
3737
        if master_oob_handler:
3738
          additional_text = ("run '%s %s %s' if you want to operate on the"
3739
                             " master regardless") % (master_oob_handler,
3740
                                                      self.op.command,
3741
                                                      self.master_node)
3742
        else:
3743
          additional_text = "it does not support out-of-band operations"
3744

    
3745
        raise errors.OpPrereqError(("Operating on the master node %s is not"
3746
                                    " allowed for %s; %s") %
3747
                                   (self.master_node, self.op.command,
3748
                                    additional_text), errors.ECODE_INVAL)
3749
    else:
3750
      self.op.node_names = self.cfg.GetNodeList()
3751
      if self.op.command in self._SKIP_MASTER:
3752
        self.op.node_names.remove(self.master_node)
3753

    
3754
    if self.op.command in self._SKIP_MASTER:
3755
      assert self.master_node not in self.op.node_names
3756

    
3757
    for node_name in self.op.node_names:
3758
      node = self.cfg.GetNodeInfo(node_name)
3759

    
3760
      if node is None:
3761
        raise errors.OpPrereqError("Node %s not found" % node_name,
3762
                                   errors.ECODE_NOENT)
3763
      else:
3764
        self.nodes.append(node)
3765

    
3766
      if (not self.op.ignore_status and
3767
          (self.op.command == constants.OOB_POWER_OFF and not node.offline)):
3768
        raise errors.OpPrereqError(("Cannot power off node %s because it is"
3769
                                    " not marked offline") % node_name,
3770
                                   errors.ECODE_STATE)
3771

    
3772
  def Exec(self, feedback_fn):
3773
    """Execute OOB and return result if we expect any.
3774

3775
    """
3776
    master_node = self.master_node
3777
    ret = []
3778

    
3779
    for idx, node in enumerate(utils.NiceSort(self.nodes,
3780
                                              key=lambda node: node.name)):
3781
      node_entry = [(constants.RS_NORMAL, node.name)]
3782
      ret.append(node_entry)
3783

    
3784
      oob_program = _SupportsOob(self.cfg, node)
3785

    
3786
      if not oob_program:
3787
        node_entry.append((constants.RS_UNAVAIL, None))
3788
        continue
3789

    
3790
      logging.info("Executing out-of-band command '%s' using '%s' on %s",
3791
                   self.op.command, oob_program, node.name)
3792
      result = self.rpc.call_run_oob(master_node, oob_program,
3793
                                     self.op.command, node.name,
3794
                                     self.op.timeout)
3795

    
3796
      if result.fail_msg:
3797
        self.LogWarning("Out-of-band RPC failed on node '%s': %s",
3798
                        node.name, result.fail_msg)
3799
        node_entry.append((constants.RS_NODATA, None))
3800
      else:
3801
        try:
3802
          self._CheckPayload(result)
3803
        except errors.OpExecError, err:
3804
          self.LogWarning("Payload returned by node '%s' is not valid: %s",
3805
                          node.name, err)
3806
          node_entry.append((constants.RS_NODATA, None))
3807
        else:
3808
          if self.op.command == constants.OOB_HEALTH:
3809
            # For health we should log important events
3810
            for item, status in result.payload:
3811
              if status in [constants.OOB_STATUS_WARNING,
3812
                            constants.OOB_STATUS_CRITICAL]:
3813
                self.LogWarning("Item '%s' on node '%s' has status '%s'",
3814
                                item, node.name, status)
3815

    
3816
          if self.op.command == constants.OOB_POWER_ON:
3817
            node.powered = True
3818
          elif self.op.command == constants.OOB_POWER_OFF:
3819
            node.powered = False
3820
          elif self.op.command == constants.OOB_POWER_STATUS:
3821
            powered = result.payload[constants.OOB_POWER_STATUS_POWERED]
3822
            if powered != node.powered:
3823
              logging.warning(("Recorded power state (%s) of node '%s' does not"
3824
                               " match actual power state (%s)"), node.powered,
3825
                              node.name, powered)
3826

    
3827
          # For configuration changing commands we should update the node
3828
          if self.op.command in (constants.OOB_POWER_ON,
3829
                                 constants.OOB_POWER_OFF):
3830
            self.cfg.Update(node, feedback_fn)
3831

    
3832
          node_entry.append((constants.RS_NORMAL, result.payload))
3833

    
3834
          if (self.op.command == constants.OOB_POWER_ON and
3835
              idx < len(self.nodes) - 1):
3836
            time.sleep(self.op.power_delay)
3837

    
3838
    return ret
3839

    
3840
  def _CheckPayload(self, result):
3841
    """Checks if the payload is valid.
3842

3843
    @param result: RPC result
3844
    @raises errors.OpExecError: If payload is not valid
3845

3846
    """
3847
    errs = []
3848
    if self.op.command == constants.OOB_HEALTH:
3849
      if not isinstance(result.payload, list):
3850
        errs.append("command 'health' is expected to return a list but got %s" %
3851
                    type(result.payload))
3852
      else:
3853
        for item, status in result.payload:
3854
          if status not in constants.OOB_STATUSES:
3855
            errs.append("health item '%s' has invalid status '%s'" %
3856
                        (item, status))
3857

    
3858
    if self.op.command == constants.OOB_POWER_STATUS:
3859
      if not isinstance(result.payload, dict):
3860
        errs.append("power-status is expected to return a dict but got %s" %
3861
                    type(result.payload))
3862

    
3863
    if self.op.command in [
3864
        constants.OOB_POWER_ON,
3865
        constants.OOB_POWER_OFF,
3866
        constants.OOB_POWER_CYCLE,
3867
        ]:
3868
      if result.payload is not None:
3869
        errs.append("%s is expected to not return payload but got '%s'" %
3870
                    (self.op.command, result.payload))
3871

    
3872
    if errs:
3873
      raise errors.OpExecError("Check of out-of-band payload failed due to %s" %
3874
                               utils.CommaJoin(errs))
3875

    
3876
class _OsQuery(_QueryBase):
3877
  FIELDS = query.OS_FIELDS
3878

    
3879
  def ExpandNames(self, lu):
3880
    # Lock all nodes in shared mode
3881
    # Temporary removal of locks, should be reverted later
3882
    # TODO: reintroduce locks when they are lighter-weight
3883
    lu.needed_locks = {}
3884
    #self.share_locks[locking.LEVEL_NODE] = 1
3885
    #self.needed_locks[locking.LEVEL_NODE] = locking.ALL_SET
3886

    
3887
    # The following variables interact with _QueryBase._GetNames
3888
    if self.names:
3889
      self.wanted = self.names
3890
    else:
3891
      self.wanted = locking.ALL_SET
3892

    
3893
    self.do_locking = self.use_locking
3894

    
3895
  def DeclareLocks(self, lu, level):
3896
    pass
3897

    
3898
  @staticmethod
3899
  def _DiagnoseByOS(rlist):
3900
    """Remaps a per-node return list into an a per-os per-node dictionary
3901

3902
    @param rlist: a map with node names as keys and OS objects as values
3903

3904
    @rtype: dict
3905
    @return: a dictionary with osnames as keys and as value another
3906
        map, with nodes as keys and tuples of (path, status, diagnose,
3907
        variants, parameters, api_versions) as values, eg::
3908

3909
          {"debian-etch": {"node1": [(/usr/lib/..., True, "", [], []),
3910
                                     (/srv/..., False, "invalid api")],
3911
                           "node2": [(/srv/..., True, "", [], [])]}
3912
          }
3913

3914
    """
3915
    all_os = {}
3916
    # we build here the list of nodes that didn't fail the RPC (at RPC
3917
    # level), so that nodes with a non-responding node daemon don't
3918
    # make all OSes invalid
3919
    good_nodes = [node_name for node_name in rlist
3920
                  if not rlist[node_name].fail_msg]
3921
    for node_name, nr in rlist.items():
3922
      if nr.fail_msg or not nr.payload:
3923
        continue
3924
      for (name, path, status, diagnose, variants,
3925
           params, api_versions) in nr.payload:
3926
        if name not in all_os:
3927
          # build a list of nodes for this os containing empty lists
3928
          # for each node in node_list
3929
          all_os[name] = {}
3930
          for nname in good_nodes:
3931
            all_os[name][nname] = []
3932
        # convert params from [name, help] to (name, help)
3933
        params = [tuple(v) for v in params]
3934
        all_os[name][node_name].append((path, status, diagnose,
3935
                                        variants, params, api_versions))
3936
    return all_os
3937

    
3938
  def _GetQueryData(self, lu):
3939
    """Computes the list of nodes and their attributes.
3940

3941
    """
3942
    # Locking is not used
3943
    assert not (compat.any(lu.glm.is_owned(level)
3944
                           for level in locking.LEVELS
3945
                           if level != locking.LEVEL_CLUSTER) or
3946
                self.do_locking or self.use_locking)
3947

    
3948
    valid_nodes = [node.name
3949
                   for node in lu.cfg.GetAllNodesInfo().values()
3950
                   if not node.offline and node.vm_capable]
3951
    pol = self._DiagnoseByOS(lu.rpc.call_os_diagnose(valid_nodes))
3952
    cluster = lu.cfg.GetClusterInfo()
3953

    
3954
    data = {}
3955

    
3956
    for (os_name, os_data) in pol.items():
3957
      info = query.OsInfo(name=os_name, valid=True, node_status=os_data,
3958
                          hidden=(os_name in cluster.hidden_os),
3959
                          blacklisted=(os_name in cluster.blacklisted_os))
3960

    
3961
      variants = set()
3962
      parameters = set()
3963
      api_versions = set()
3964

    
3965
      for idx, osl in enumerate(os_data.values()):
3966
        info.valid = bool(info.valid and osl and osl[0][1])
3967
        if not info.valid:
3968
          break
3969

    
3970
        (node_variants, node_params, node_api) = osl[0][3:6]
3971
        if idx == 0:
3972
          # First entry
3973
          variants.update(node_variants)
3974
          parameters.update(node_params)
3975
          api_versions.update(node_api)
3976
        else:
3977
          # Filter out inconsistent values
3978
          variants.intersection_update(node_variants)
3979
          parameters.intersection_update(node_params)
3980
          api_versions.intersection_update(node_api)
3981

    
3982
      info.variants = list(variants)
3983
      info.parameters = list(parameters)
3984
      info.api_versions = list(api_versions)
3985

    
3986
      data[os_name] = info
3987

    
3988
    # Prepare data in requested order
3989
    return [data[name] for name in self._GetNames(lu, pol.keys(), None)
3990
            if name in data]
3991

    
3992

    
3993
class LUOsDiagnose(NoHooksLU):
3994
  """Logical unit for OS diagnose/query.
3995

3996
  """
3997
  REQ_BGL = False
3998

    
3999
  @staticmethod
4000
  def _BuildFilter(fields, names):
4001
    """Builds a filter for querying OSes.
4002

4003
    """
4004
    name_filter = qlang.MakeSimpleFilter("name", names)
4005

    
4006
    # Legacy behaviour: Hide hidden, blacklisted or invalid OSes if the
4007
    # respective field is not requested
4008
    status_filter = [[qlang.OP_NOT, [qlang.OP_TRUE, fname]]
4009
                     for fname in ["hidden", "blacklisted"]
4010
                     if fname not in fields]
4011
    if "valid" not in fields:
4012
      status_filter.append([qlang.OP_TRUE, "valid"])
4013

    
4014
    if status_filter:
4015
      status_filter.insert(0, qlang.OP_AND)
4016
    else:
4017
      status_filter = None
4018

    
4019
    if name_filter and status_filter:
4020
      return [qlang.OP_AND, name_filter, status_filter]
4021
    elif name_filter:
4022
      return name_filter
4023
    else:
4024
      return status_filter
4025

    
4026
  def CheckArguments(self):
4027
    self.oq = _OsQuery(self._BuildFilter(self.op.output_fields, self.op.names),
4028
                       self.op.output_fields, False)
4029

    
4030
  def ExpandNames(self):
4031
    self.oq.ExpandNames(self)
4032

    
4033
  def Exec(self, feedback_fn):
4034
    return self.oq.OldStyleQuery(self)
4035

    
4036

    
4037
class LUNodeRemove(LogicalUnit):
4038
  """Logical unit for removing a node.
4039

4040
  """
4041
  HPATH = "node-remove"
4042
  HTYPE = constants.HTYPE_NODE
4043

    
4044
  def BuildHooksEnv(self):
4045
    """Build hooks env.
4046

4047
    This doesn't run on the target node in the pre phase as a failed
4048
    node would then be impossible to remove.
4049

4050
    """
4051
    return {
4052
      "OP_TARGET": self.op.node_name,
4053
      "NODE_NAME": self.op.node_name,
4054
      }
4055

    
4056
  def BuildHooksNodes(self):
4057
    """Build hooks nodes.
4058

4059
    """
4060
    all_nodes = self.cfg.GetNodeList()
4061
    try:
4062
      all_nodes.remove(self.op.node_name)
4063
    except ValueError:
4064
      logging.warning("Node '%s', which is about to be removed, was not found"
4065
                      " in the list of all nodes", self.op.node_name)
4066
    return (all_nodes, all_nodes)
4067

    
4068
  def CheckPrereq(self):
4069
    """Check prerequisites.
4070

4071
    This checks: