Statistics
| Branch: | Tag: | Revision:

root / lib / cmdlib.py @ 5d55819e

History | View | Annotate | Download (322.3 kB)

1
#
2
#
3

    
4
# Copyright (C) 2006, 2007, 2008 Google Inc.
5
#
6
# This program is free software; you can redistribute it and/or modify
7
# it under the terms of the GNU General Public License as published by
8
# the Free Software Foundation; either version 2 of the License, or
9
# (at your option) any later version.
10
#
11
# This program is distributed in the hope that it will be useful, but
12
# WITHOUT ANY WARRANTY; without even the implied warranty of
13
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
14
# General Public License for more details.
15
#
16
# You should have received a copy of the GNU General Public License
17
# along with this program; if not, write to the Free Software
18
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
19
# 02110-1301, USA.
20

    
21

    
22
"""Module implementing the master-side code."""
23

    
24
# pylint: disable-msg=W0201
25

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

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

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

    
48

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

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

61
  Note that all commands require root permissions.
62

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

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

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

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

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

    
104
    # Tasklets
105
    self.tasklets = None
106

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

    
113
    self.CheckArguments()
114

    
115
  def __GetSSH(self):
116
    """Returns the SshRunner object
117

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

    
123
  ssh = property(fget=__GetSSH)
124

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

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

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

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

140
    """
141
    pass
142

    
143
  def ExpandNames(self):
144
    """Expand names for this LU.
145

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

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

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

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

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

168
    Examples::
169

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

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

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

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

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

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

207
    """
208

    
209
  def CheckPrereq(self):
210
    """Check prerequisites for this LU.
211

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

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

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

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

    
232
  def Exec(self, feedback_fn):
233
    """Execute the LU.
234

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

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

    
247
  def BuildHooksEnv(self):
248
    """Build hooks environment for this LU.
249

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

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

261
    No nodes should be returned as an empty list (and not None).
262

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

266
    """
267
    raise NotImplementedError
268

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

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

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

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

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

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

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

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

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

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

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

326
    If should be called in DeclareLocks in a way similar to::
327

328
      if level == locking.LEVEL_NODE:
329
        self._LockInstancesNodes()
330

331
    @type primary_only: boolean
332
    @param primary_only: only lock primary nodes of locked instances
333

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

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

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

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

    
355
    del self.recalculate_locks[locking.LEVEL_NODE]
356

    
357

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

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

364
  """
365
  HPATH = None
366
  HTYPE = None
367

    
368
  def BuildHooksEnv(self):
369
    """Empty BuildHooksEnv for NoHooksLu.
370

371
    This just raises an error.
372

373
    """
374
    assert False, "BuildHooksEnv called for NoHooksLUs"
375

    
376

    
377
class Tasklet:
378
  """Tasklet base class.
379

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

384
  Subclasses must follow these rules:
385
    - Implement CheckPrereq
386
    - Implement Exec
387

388
  """
389
  def __init__(self, lu):
390
    self.lu = lu
391

    
392
    # Shortcuts
393
    self.cfg = lu.cfg
394
    self.rpc = lu.rpc
395

    
396
  def CheckPrereq(self):
397
    """Check prerequisites for this tasklets.
398

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

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

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

409
    """
410
    raise NotImplementedError
411

    
412
  def Exec(self, feedback_fn):
413
    """Execute the tasklet.
414

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

419
    """
420
    raise NotImplementedError
421

    
422

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

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

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

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

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

    
446

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

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

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

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

    
470

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

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

479
  """
480
  f = utils.FieldSet()
481
  f.Extend(static)
482
  f.Extend(dynamic)
483

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

    
489

    
490
def _CheckBooleanOpField(op, name):
491
  """Validates boolean opcode parameters.
492

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

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

    
503

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

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

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

    
518

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

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

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

    
531

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

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

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

    
544

    
545
def _CheckDiskTemplate(template):
546
  """Ensure a given disk template is valid.
547

548
  """
549
  if template not in constants.DISK_TEMPLATES:
550
    msg = ("Invalid disk template name '%s', valid templates are: %s" %
551
           (template, utils.CommaJoin(constants.DISK_TEMPLATES)))
552
    raise errors.OpPrereqError(msg, errors.ECODE_INVAL)
553

    
554

    
555
def _ExpandItemName(fn, name, kind):
556
  """Expand an item name.
557

558
  @param fn: the function to use for expansion
559
  @param name: requested item name
560
  @param kind: text description ('Node' or 'Instance')
561
  @return: the resolved (full) name
562
  @raise errors.OpPrereqError: if the item is not found
563

564
  """
565
  full_name = fn(name)
566
  if full_name is None:
567
    raise errors.OpPrereqError("%s '%s' not known" % (kind, name),
568
                               errors.ECODE_NOENT)
569
  return full_name
570

    
571

    
572
def _ExpandNodeName(cfg, name):
573
  """Wrapper over L{_ExpandItemName} for nodes."""
574
  return _ExpandItemName(cfg.ExpandNodeName, name, "Node")
575

    
576

    
577
def _ExpandInstanceName(cfg, name):
578
  """Wrapper over L{_ExpandItemName} for instance."""
579
  return _ExpandItemName(cfg.ExpandInstanceName, name, "Instance")
580

    
581

    
582
def _BuildInstanceHookEnv(name, primary_node, secondary_nodes, os_type, status,
583
                          memory, vcpus, nics, disk_template, disks,
584
                          bep, hvp, hypervisor_name):
585
  """Builds instance related env variables for hooks
586

587
  This builds the hook environment from individual variables.
588

589
  @type name: string
590
  @param name: the name of the instance
591
  @type primary_node: string
592
  @param primary_node: the name of the instance's primary node
593
  @type secondary_nodes: list
594
  @param secondary_nodes: list of secondary nodes as strings
595
  @type os_type: string
596
  @param os_type: the name of the instance's OS
597
  @type status: boolean
598
  @param status: the should_run status of the instance
599
  @type memory: string
600
  @param memory: the memory size of the instance
601
  @type vcpus: string
602
  @param vcpus: the count of VCPUs the instance has
603
  @type nics: list
604
  @param nics: list of tuples (ip, mac, mode, link) representing
605
      the NICs the instance has
606
  @type disk_template: string
607
  @param disk_template: the disk template of the instance
608
  @type disks: list
609
  @param disks: the list of (size, mode) pairs
610
  @type bep: dict
611
  @param bep: the backend parameters for the instance
612
  @type hvp: dict
613
  @param hvp: the hypervisor parameters for the instance
614
  @type hypervisor_name: string
615
  @param hypervisor_name: the hypervisor for the instance
616
  @rtype: dict
617
  @return: the hook environment for this instance
618

619
  """
620
  if status:
621
    str_status = "up"
622
  else:
623
    str_status = "down"
624
  env = {
625
    "OP_TARGET": name,
626
    "INSTANCE_NAME": name,
627
    "INSTANCE_PRIMARY": primary_node,
628
    "INSTANCE_SECONDARIES": " ".join(secondary_nodes),
629
    "INSTANCE_OS_TYPE": os_type,
630
    "INSTANCE_STATUS": str_status,
631
    "INSTANCE_MEMORY": memory,
632
    "INSTANCE_VCPUS": vcpus,
633
    "INSTANCE_DISK_TEMPLATE": disk_template,
634
    "INSTANCE_HYPERVISOR": hypervisor_name,
635
  }
636

    
637
  if nics:
638
    nic_count = len(nics)
639
    for idx, (ip, mac, mode, link) in enumerate(nics):
640
      if ip is None:
641
        ip = ""
642
      env["INSTANCE_NIC%d_IP" % idx] = ip
643
      env["INSTANCE_NIC%d_MAC" % idx] = mac
644
      env["INSTANCE_NIC%d_MODE" % idx] = mode
645
      env["INSTANCE_NIC%d_LINK" % idx] = link
646
      if mode == constants.NIC_MODE_BRIDGED:
647
        env["INSTANCE_NIC%d_BRIDGE" % idx] = link
648
  else:
649
    nic_count = 0
650

    
651
  env["INSTANCE_NIC_COUNT"] = nic_count
652

    
653
  if disks:
654
    disk_count = len(disks)
655
    for idx, (size, mode) in enumerate(disks):
656
      env["INSTANCE_DISK%d_SIZE" % idx] = size
657
      env["INSTANCE_DISK%d_MODE" % idx] = mode
658
  else:
659
    disk_count = 0
660

    
661
  env["INSTANCE_DISK_COUNT"] = disk_count
662

    
663
  for source, kind in [(bep, "BE"), (hvp, "HV")]:
664
    for key, value in source.items():
665
      env["INSTANCE_%s_%s" % (kind, key)] = value
666

    
667
  return env
668

    
669

    
670
def _NICListToTuple(lu, nics):
671
  """Build a list of nic information tuples.
672

673
  This list is suitable to be passed to _BuildInstanceHookEnv or as a return
674
  value in LUQueryInstanceData.
675

676
  @type lu:  L{LogicalUnit}
677
  @param lu: the logical unit on whose behalf we execute
678
  @type nics: list of L{objects.NIC}
679
  @param nics: list of nics to convert to hooks tuples
680

681
  """
682
  hooks_nics = []
683
  c_nicparams = lu.cfg.GetClusterInfo().nicparams[constants.PP_DEFAULT]
684
  for nic in nics:
685
    ip = nic.ip
686
    mac = nic.mac
687
    filled_params = objects.FillDict(c_nicparams, nic.nicparams)
688
    mode = filled_params[constants.NIC_MODE]
689
    link = filled_params[constants.NIC_LINK]
690
    hooks_nics.append((ip, mac, mode, link))
691
  return hooks_nics
692

    
693

    
694
def _BuildInstanceHookEnvByObject(lu, instance, override=None):
695
  """Builds instance related env variables for hooks from an object.
696

697
  @type lu: L{LogicalUnit}
698
  @param lu: the logical unit on whose behalf we execute
699
  @type instance: L{objects.Instance}
700
  @param instance: the instance for which we should build the
701
      environment
702
  @type override: dict
703
  @param override: dictionary with key/values that will override
704
      our values
705
  @rtype: dict
706
  @return: the hook environment dictionary
707

708
  """
709
  cluster = lu.cfg.GetClusterInfo()
710
  bep = cluster.FillBE(instance)
711
  hvp = cluster.FillHV(instance)
712
  args = {
713
    'name': instance.name,
714
    'primary_node': instance.primary_node,
715
    'secondary_nodes': instance.secondary_nodes,
716
    'os_type': instance.os,
717
    'status': instance.admin_up,
718
    'memory': bep[constants.BE_MEMORY],
719
    'vcpus': bep[constants.BE_VCPUS],
720
    'nics': _NICListToTuple(lu, instance.nics),
721
    'disk_template': instance.disk_template,
722
    'disks': [(disk.size, disk.mode) for disk in instance.disks],
723
    'bep': bep,
724
    'hvp': hvp,
725
    'hypervisor_name': instance.hypervisor,
726
  }
727
  if override:
728
    args.update(override)
729
  return _BuildInstanceHookEnv(**args) # pylint: disable-msg=W0142
730

    
731

    
732
def _AdjustCandidatePool(lu, exceptions):
733
  """Adjust the candidate pool after node operations.
734

735
  """
736
  mod_list = lu.cfg.MaintainCandidatePool(exceptions)
737
  if mod_list:
738
    lu.LogInfo("Promoted nodes to master candidate role: %s",
739
               utils.CommaJoin(node.name for node in mod_list))
740
    for name in mod_list:
741
      lu.context.ReaddNode(name)
742
  mc_now, mc_max, _ = lu.cfg.GetMasterCandidateStats(exceptions)
743
  if mc_now > mc_max:
744
    lu.LogInfo("Note: more nodes are candidates (%d) than desired (%d)" %
745
               (mc_now, mc_max))
746

    
747

    
748
def _DecideSelfPromotion(lu, exceptions=None):
749
  """Decide whether I should promote myself as a master candidate.
750

751
  """
752
  cp_size = lu.cfg.GetClusterInfo().candidate_pool_size
753
  mc_now, mc_should, _ = lu.cfg.GetMasterCandidateStats(exceptions)
754
  # the new node will increase mc_max with one, so:
755
  mc_should = min(mc_should + 1, cp_size)
756
  return mc_now < mc_should
757

    
758

    
759
def _CheckNicsBridgesExist(lu, target_nics, target_node,
760
                               profile=constants.PP_DEFAULT):
761
  """Check that the brigdes needed by a list of nics exist.
762

763
  """
764
  c_nicparams = lu.cfg.GetClusterInfo().nicparams[profile]
765
  paramslist = [objects.FillDict(c_nicparams, nic.nicparams)
766
                for nic in target_nics]
767
  brlist = [params[constants.NIC_LINK] for params in paramslist
768
            if params[constants.NIC_MODE] == constants.NIC_MODE_BRIDGED]
769
  if brlist:
770
    result = lu.rpc.call_bridges_exist(target_node, brlist)
771
    result.Raise("Error checking bridges on destination node '%s'" %
772
                 target_node, prereq=True, ecode=errors.ECODE_ENVIRON)
773

    
774

    
775
def _CheckInstanceBridgesExist(lu, instance, node=None):
776
  """Check that the brigdes needed by an instance exist.
777

778
  """
779
  if node is None:
780
    node = instance.primary_node
781
  _CheckNicsBridgesExist(lu, instance.nics, node)
782

    
783

    
784
def _CheckOSVariant(os_obj, name):
785
  """Check whether an OS name conforms to the os variants specification.
786

787
  @type os_obj: L{objects.OS}
788
  @param os_obj: OS object to check
789
  @type name: string
790
  @param name: OS name passed by the user, to check for validity
791

792
  """
793
  if not os_obj.supported_variants:
794
    return
795
  try:
796
    variant = name.split("+", 1)[1]
797
  except IndexError:
798
    raise errors.OpPrereqError("OS name must include a variant",
799
                               errors.ECODE_INVAL)
800

    
801
  if variant not in os_obj.supported_variants:
802
    raise errors.OpPrereqError("Unsupported OS variant", errors.ECODE_INVAL)
803

    
804

    
805
def _GetNodeInstancesInner(cfg, fn):
806
  return [i for i in cfg.GetAllInstancesInfo().values() if fn(i)]
807

    
808

    
809
def _GetNodeInstances(cfg, node_name):
810
  """Returns a list of all primary and secondary instances on a node.
811

812
  """
813

    
814
  return _GetNodeInstancesInner(cfg, lambda inst: node_name in inst.all_nodes)
815

    
816

    
817
def _GetNodePrimaryInstances(cfg, node_name):
818
  """Returns primary instances on a node.
819

820
  """
821
  return _GetNodeInstancesInner(cfg,
822
                                lambda inst: node_name == inst.primary_node)
823

    
824

    
825
def _GetNodeSecondaryInstances(cfg, node_name):
826
  """Returns secondary instances on a node.
827

828
  """
829
  return _GetNodeInstancesInner(cfg,
830
                                lambda inst: node_name in inst.secondary_nodes)
831

    
832

    
833
def _GetStorageTypeArgs(cfg, storage_type):
834
  """Returns the arguments for a storage type.
835

836
  """
837
  # Special case for file storage
838
  if storage_type == constants.ST_FILE:
839
    # storage.FileStorage wants a list of storage directories
840
    return [[cfg.GetFileStorageDir()]]
841

    
842
  return []
843

    
844

    
845
def _FindFaultyInstanceDisks(cfg, rpc, instance, node_name, prereq):
846
  faulty = []
847

    
848
  for dev in instance.disks:
849
    cfg.SetDiskID(dev, node_name)
850

    
851
  result = rpc.call_blockdev_getmirrorstatus(node_name, instance.disks)
852
  result.Raise("Failed to get disk status from node %s" % node_name,
853
               prereq=prereq, ecode=errors.ECODE_ENVIRON)
854

    
855
  for idx, bdev_status in enumerate(result.payload):
856
    if bdev_status and bdev_status.ldisk_status == constants.LDS_FAULTY:
857
      faulty.append(idx)
858

    
859
  return faulty
860

    
861

    
862
def _FormatTimestamp(secs):
863
  """Formats a Unix timestamp with the local timezone.
864

865
  """
866
  return time.strftime("%F %T %Z", time.gmtime(secs))
867

    
868

    
869
class LUPostInitCluster(LogicalUnit):
870
  """Logical unit for running hooks after cluster initialization.
871

872
  """
873
  HPATH = "cluster-init"
874
  HTYPE = constants.HTYPE_CLUSTER
875
  _OP_REQP = []
876

    
877
  def BuildHooksEnv(self):
878
    """Build hooks env.
879

880
    """
881
    env = {"OP_TARGET": self.cfg.GetClusterName()}
882
    mn = self.cfg.GetMasterNode()
883
    return env, [], [mn]
884

    
885
  def CheckPrereq(self):
886
    """No prerequisites to check.
887

888
    """
889
    return True
890

    
891
  def Exec(self, feedback_fn):
892
    """Nothing to do.
893

894
    """
895
    return True
896

    
897

    
898
class LUDestroyCluster(LogicalUnit):
899
  """Logical unit for destroying the cluster.
900

901
  """
902
  HPATH = "cluster-destroy"
903
  HTYPE = constants.HTYPE_CLUSTER
904
  _OP_REQP = []
905

    
906
  def BuildHooksEnv(self):
907
    """Build hooks env.
908

909
    """
910
    env = {"OP_TARGET": self.cfg.GetClusterName()}
911
    return env, [], []
912

    
913
  def CheckPrereq(self):
914
    """Check prerequisites.
915

916
    This checks whether the cluster is empty.
917

918
    Any errors are signaled by raising errors.OpPrereqError.
919

920
    """
921
    master = self.cfg.GetMasterNode()
922

    
923
    nodelist = self.cfg.GetNodeList()
924
    if len(nodelist) != 1 or nodelist[0] != master:
925
      raise errors.OpPrereqError("There are still %d node(s) in"
926
                                 " this cluster." % (len(nodelist) - 1),
927
                                 errors.ECODE_INVAL)
928
    instancelist = self.cfg.GetInstanceList()
929
    if instancelist:
930
      raise errors.OpPrereqError("There are still %d instance(s) in"
931
                                 " this cluster." % len(instancelist),
932
                                 errors.ECODE_INVAL)
933

    
934
  def Exec(self, feedback_fn):
935
    """Destroys the cluster.
936

937
    """
938
    master = self.cfg.GetMasterNode()
939
    modify_ssh_setup = self.cfg.GetClusterInfo().modify_ssh_setup
940

    
941
    # Run post hooks on master node before it's removed
942
    hm = self.proc.hmclass(self.rpc.call_hooks_runner, self)
943
    try:
944
      hm.RunPhase(constants.HOOKS_PHASE_POST, [master])
945
    except:
946
      # pylint: disable-msg=W0702
947
      self.LogWarning("Errors occurred running hooks on %s" % master)
948

    
949
    result = self.rpc.call_node_stop_master(master, False)
950
    result.Raise("Could not disable the master role")
951

    
952
    if modify_ssh_setup:
953
      priv_key, pub_key, _ = ssh.GetUserFiles(constants.GANETI_RUNAS)
954
      utils.CreateBackup(priv_key)
955
      utils.CreateBackup(pub_key)
956

    
957
    return master
958

    
959

    
960
def _VerifyCertificateInner(filename, expired, not_before, not_after, now,
961
                            warn_days=constants.SSL_CERT_EXPIRATION_WARN,
962
                            error_days=constants.SSL_CERT_EXPIRATION_ERROR):
963
  """Verifies certificate details for LUVerifyCluster.
964

965
  """
966
  if expired:
967
    msg = "Certificate %s is expired" % filename
968

    
969
    if not_before is not None and not_after is not None:
970
      msg += (" (valid from %s to %s)" %
971
              (_FormatTimestamp(not_before),
972
               _FormatTimestamp(not_after)))
973
    elif not_before is not None:
974
      msg += " (valid from %s)" % _FormatTimestamp(not_before)
975
    elif not_after is not None:
976
      msg += " (valid until %s)" % _FormatTimestamp(not_after)
977

    
978
    return (LUVerifyCluster.ETYPE_ERROR, msg)
979

    
980
  elif not_before is not None and not_before > now:
981
    return (LUVerifyCluster.ETYPE_WARNING,
982
            "Certificate %s not yet valid (valid from %s)" %
983
            (filename, _FormatTimestamp(not_before)))
984

    
985
  elif not_after is not None:
986
    remaining_days = int((not_after - now) / (24 * 3600))
987

    
988
    msg = ("Certificate %s expires in %d days" % (filename, remaining_days))
989

    
990
    if remaining_days <= error_days:
991
      return (LUVerifyCluster.ETYPE_ERROR, msg)
992

    
993
    if remaining_days <= warn_days:
994
      return (LUVerifyCluster.ETYPE_WARNING, msg)
995

    
996
  return (None, None)
997

    
998

    
999
def _VerifyCertificate(filename):
1000
  """Verifies a certificate for LUVerifyCluster.
1001

1002
  @type filename: string
1003
  @param filename: Path to PEM file
1004

1005
  """
1006
  try:
1007
    cert = OpenSSL.crypto.load_certificate(OpenSSL.crypto.FILETYPE_PEM,
1008
                                           utils.ReadFile(filename))
1009
  except Exception, err: # pylint: disable-msg=W0703
1010
    return (LUVerifyCluster.ETYPE_ERROR,
1011
            "Failed to load X509 certificate %s: %s" % (filename, err))
1012

    
1013
  # Depending on the pyOpenSSL version, this can just return (None, None)
1014
  (not_before, not_after) = utils.GetX509CertValidity(cert)
1015

    
1016
  return _VerifyCertificateInner(filename, cert.has_expired(),
1017
                                 not_before, not_after, time.time())
1018

    
1019

    
1020
class LUVerifyCluster(LogicalUnit):
1021
  """Verifies the cluster status.
1022

1023
  """
1024
  HPATH = "cluster-verify"
1025
  HTYPE = constants.HTYPE_CLUSTER
1026
  _OP_REQP = ["skip_checks", "verbose", "error_codes", "debug_simulate_errors"]
1027
  REQ_BGL = False
1028

    
1029
  TCLUSTER = "cluster"
1030
  TNODE = "node"
1031
  TINSTANCE = "instance"
1032

    
1033
  ECLUSTERCFG = (TCLUSTER, "ECLUSTERCFG")
1034
  ECLUSTERCERT = (TCLUSTER, "ECLUSTERCERT")
1035
  EINSTANCEBADNODE = (TINSTANCE, "EINSTANCEBADNODE")
1036
  EINSTANCEDOWN = (TINSTANCE, "EINSTANCEDOWN")
1037
  EINSTANCELAYOUT = (TINSTANCE, "EINSTANCELAYOUT")
1038
  EINSTANCEMISSINGDISK = (TINSTANCE, "EINSTANCEMISSINGDISK")
1039
  EINSTANCEMISSINGDISK = (TINSTANCE, "EINSTANCEMISSINGDISK")
1040
  EINSTANCEWRONGNODE = (TINSTANCE, "EINSTANCEWRONGNODE")
1041
  ENODEDRBD = (TNODE, "ENODEDRBD")
1042
  ENODEFILECHECK = (TNODE, "ENODEFILECHECK")
1043
  ENODEHOOKS = (TNODE, "ENODEHOOKS")
1044
  ENODEHV = (TNODE, "ENODEHV")
1045
  ENODELVM = (TNODE, "ENODELVM")
1046
  ENODEN1 = (TNODE, "ENODEN1")
1047
  ENODENET = (TNODE, "ENODENET")
1048
  ENODEORPHANINSTANCE = (TNODE, "ENODEORPHANINSTANCE")
1049
  ENODEORPHANLV = (TNODE, "ENODEORPHANLV")
1050
  ENODERPC = (TNODE, "ENODERPC")
1051
  ENODESSH = (TNODE, "ENODESSH")
1052
  ENODEVERSION = (TNODE, "ENODEVERSION")
1053
  ENODESETUP = (TNODE, "ENODESETUP")
1054
  ENODETIME = (TNODE, "ENODETIME")
1055

    
1056
  ETYPE_FIELD = "code"
1057
  ETYPE_ERROR = "ERROR"
1058
  ETYPE_WARNING = "WARNING"
1059

    
1060
  def ExpandNames(self):
1061
    self.needed_locks = {
1062
      locking.LEVEL_NODE: locking.ALL_SET,
1063
      locking.LEVEL_INSTANCE: locking.ALL_SET,
1064
    }
1065
    self.share_locks = dict.fromkeys(locking.LEVELS, 1)
1066

    
1067
  def _Error(self, ecode, item, msg, *args, **kwargs):
1068
    """Format an error message.
1069

1070
    Based on the opcode's error_codes parameter, either format a
1071
    parseable error code, or a simpler error string.
1072

1073
    This must be called only from Exec and functions called from Exec.
1074

1075
    """
1076
    ltype = kwargs.get(self.ETYPE_FIELD, self.ETYPE_ERROR)
1077
    itype, etxt = ecode
1078
    # first complete the msg
1079
    if args:
1080
      msg = msg % args
1081
    # then format the whole message
1082
    if self.op.error_codes:
1083
      msg = "%s:%s:%s:%s:%s" % (ltype, etxt, itype, item, msg)
1084
    else:
1085
      if item:
1086
        item = " " + item
1087
      else:
1088
        item = ""
1089
      msg = "%s: %s%s: %s" % (ltype, itype, item, msg)
1090
    # and finally report it via the feedback_fn
1091
    self._feedback_fn("  - %s" % msg)
1092

    
1093
  def _ErrorIf(self, cond, *args, **kwargs):
1094
    """Log an error message if the passed condition is True.
1095

1096
    """
1097
    cond = bool(cond) or self.op.debug_simulate_errors
1098
    if cond:
1099
      self._Error(*args, **kwargs)
1100
    # do not mark the operation as failed for WARN cases only
1101
    if kwargs.get(self.ETYPE_FIELD, self.ETYPE_ERROR) == self.ETYPE_ERROR:
1102
      self.bad = self.bad or cond
1103

    
1104
  def _VerifyNode(self, nodeinfo, file_list, local_cksum,
1105
                  node_result, master_files, drbd_map, vg_name):
1106
    """Run multiple tests against a node.
1107

1108
    Test list:
1109

1110
      - compares ganeti version
1111
      - checks vg existence and size > 20G
1112
      - checks config file checksum
1113
      - checks ssh to other nodes
1114

1115
    @type nodeinfo: L{objects.Node}
1116
    @param nodeinfo: the node to check
1117
    @param file_list: required list of files
1118
    @param local_cksum: dictionary of local files and their checksums
1119
    @param node_result: the results from the node
1120
    @param master_files: list of files that only masters should have
1121
    @param drbd_map: the useddrbd minors for this node, in
1122
        form of minor: (instance, must_exist) which correspond to instances
1123
        and their running status
1124
    @param vg_name: Ganeti Volume Group (result of self.cfg.GetVGName())
1125

1126
    """
1127
    node = nodeinfo.name
1128
    _ErrorIf = self._ErrorIf # pylint: disable-msg=C0103
1129

    
1130
    # main result, node_result should be a non-empty dict
1131
    test = not node_result or not isinstance(node_result, dict)
1132
    _ErrorIf(test, self.ENODERPC, node,
1133
                  "unable to verify node: no data returned")
1134
    if test:
1135
      return
1136

    
1137
    # compares ganeti version
1138
    local_version = constants.PROTOCOL_VERSION
1139
    remote_version = node_result.get('version', None)
1140
    test = not (remote_version and
1141
                isinstance(remote_version, (list, tuple)) and
1142
                len(remote_version) == 2)
1143
    _ErrorIf(test, self.ENODERPC, node,
1144
             "connection to node returned invalid data")
1145
    if test:
1146
      return
1147

    
1148
    test = local_version != remote_version[0]
1149
    _ErrorIf(test, self.ENODEVERSION, node,
1150
             "incompatible protocol versions: master %s,"
1151
             " node %s", local_version, remote_version[0])
1152
    if test:
1153
      return
1154

    
1155
    # node seems compatible, we can actually try to look into its results
1156

    
1157
    # full package version
1158
    self._ErrorIf(constants.RELEASE_VERSION != remote_version[1],
1159
                  self.ENODEVERSION, node,
1160
                  "software version mismatch: master %s, node %s",
1161
                  constants.RELEASE_VERSION, remote_version[1],
1162
                  code=self.ETYPE_WARNING)
1163

    
1164
    # checks vg existence and size > 20G
1165
    if vg_name is not None:
1166
      vglist = node_result.get(constants.NV_VGLIST, None)
1167
      test = not vglist
1168
      _ErrorIf(test, self.ENODELVM, node, "unable to check volume groups")
1169
      if not test:
1170
        vgstatus = utils.CheckVolumeGroupSize(vglist, vg_name,
1171
                                              constants.MIN_VG_SIZE)
1172
        _ErrorIf(vgstatus, self.ENODELVM, node, vgstatus)
1173

    
1174
    # checks config file checksum
1175

    
1176
    remote_cksum = node_result.get(constants.NV_FILELIST, None)
1177
    test = not isinstance(remote_cksum, dict)
1178
    _ErrorIf(test, self.ENODEFILECHECK, node,
1179
             "node hasn't returned file checksum data")
1180
    if not test:
1181
      for file_name in file_list:
1182
        node_is_mc = nodeinfo.master_candidate
1183
        must_have = (file_name not in master_files) or node_is_mc
1184
        # missing
1185
        test1 = file_name not in remote_cksum
1186
        # invalid checksum
1187
        test2 = not test1 and remote_cksum[file_name] != local_cksum[file_name]
1188
        # existing and good
1189
        test3 = not test1 and remote_cksum[file_name] == local_cksum[file_name]
1190
        _ErrorIf(test1 and must_have, self.ENODEFILECHECK, node,
1191
                 "file '%s' missing", file_name)
1192
        _ErrorIf(test2 and must_have, self.ENODEFILECHECK, node,
1193
                 "file '%s' has wrong checksum", file_name)
1194
        # not candidate and this is not a must-have file
1195
        _ErrorIf(test2 and not must_have, self.ENODEFILECHECK, node,
1196
                 "file '%s' should not exist on non master"
1197
                 " candidates (and the file is outdated)", file_name)
1198
        # all good, except non-master/non-must have combination
1199
        _ErrorIf(test3 and not must_have, self.ENODEFILECHECK, node,
1200
                 "file '%s' should not exist"
1201
                 " on non master candidates", file_name)
1202

    
1203
    # checks ssh to any
1204

    
1205
    test = constants.NV_NODELIST not in node_result
1206
    _ErrorIf(test, self.ENODESSH, node,
1207
             "node hasn't returned node ssh connectivity data")
1208
    if not test:
1209
      if node_result[constants.NV_NODELIST]:
1210
        for a_node, a_msg in node_result[constants.NV_NODELIST].items():
1211
          _ErrorIf(True, self.ENODESSH, node,
1212
                   "ssh communication with node '%s': %s", a_node, a_msg)
1213

    
1214
    test = constants.NV_NODENETTEST not in node_result
1215
    _ErrorIf(test, self.ENODENET, node,
1216
             "node hasn't returned node tcp connectivity data")
1217
    if not test:
1218
      if node_result[constants.NV_NODENETTEST]:
1219
        nlist = utils.NiceSort(node_result[constants.NV_NODENETTEST].keys())
1220
        for anode in nlist:
1221
          _ErrorIf(True, self.ENODENET, node,
1222
                   "tcp communication with node '%s': %s",
1223
                   anode, node_result[constants.NV_NODENETTEST][anode])
1224

    
1225
    hyp_result = node_result.get(constants.NV_HYPERVISOR, None)
1226
    if isinstance(hyp_result, dict):
1227
      for hv_name, hv_result in hyp_result.iteritems():
1228
        test = hv_result is not None
1229
        _ErrorIf(test, self.ENODEHV, node,
1230
                 "hypervisor %s verify failure: '%s'", hv_name, hv_result)
1231

    
1232
    # check used drbd list
1233
    if vg_name is not None:
1234
      used_minors = node_result.get(constants.NV_DRBDLIST, [])
1235
      test = not isinstance(used_minors, (tuple, list))
1236
      _ErrorIf(test, self.ENODEDRBD, node,
1237
               "cannot parse drbd status file: %s", str(used_minors))
1238
      if not test:
1239
        for minor, (iname, must_exist) in drbd_map.items():
1240
          test = minor not in used_minors and must_exist
1241
          _ErrorIf(test, self.ENODEDRBD, node,
1242
                   "drbd minor %d of instance %s is not active",
1243
                   minor, iname)
1244
        for minor in used_minors:
1245
          test = minor not in drbd_map
1246
          _ErrorIf(test, self.ENODEDRBD, node,
1247
                   "unallocated drbd minor %d is in use", minor)
1248
    test = node_result.get(constants.NV_NODESETUP,
1249
                           ["Missing NODESETUP results"])
1250
    _ErrorIf(test, self.ENODESETUP, node, "node setup error: %s",
1251
             "; ".join(test))
1252

    
1253
    # check pv names
1254
    if vg_name is not None:
1255
      pvlist = node_result.get(constants.NV_PVLIST, None)
1256
      test = pvlist is None
1257
      _ErrorIf(test, self.ENODELVM, node, "Can't get PV list from node")
1258
      if not test:
1259
        # check that ':' is not present in PV names, since it's a
1260
        # special character for lvcreate (denotes the range of PEs to
1261
        # use on the PV)
1262
        for _, pvname, owner_vg in pvlist:
1263
          test = ":" in pvname
1264
          _ErrorIf(test, self.ENODELVM, node, "Invalid character ':' in PV"
1265
                   " '%s' of VG '%s'", pvname, owner_vg)
1266

    
1267
  def _VerifyInstance(self, instance, instanceconfig, node_vol_is,
1268
                      node_instance, n_offline):
1269
    """Verify an instance.
1270

1271
    This function checks to see if the required block devices are
1272
    available on the instance's node.
1273

1274
    """
1275
    _ErrorIf = self._ErrorIf # pylint: disable-msg=C0103
1276
    node_current = instanceconfig.primary_node
1277

    
1278
    node_vol_should = {}
1279
    instanceconfig.MapLVsByNode(node_vol_should)
1280

    
1281
    for node in node_vol_should:
1282
      if node in n_offline:
1283
        # ignore missing volumes on offline nodes
1284
        continue
1285
      for volume in node_vol_should[node]:
1286
        test = node not in node_vol_is or volume not in node_vol_is[node]
1287
        _ErrorIf(test, self.EINSTANCEMISSINGDISK, instance,
1288
                 "volume %s missing on node %s", volume, node)
1289

    
1290
    if instanceconfig.admin_up:
1291
      test = ((node_current not in node_instance or
1292
               not instance in node_instance[node_current]) and
1293
              node_current not in n_offline)
1294
      _ErrorIf(test, self.EINSTANCEDOWN, instance,
1295
               "instance not running on its primary node %s",
1296
               node_current)
1297

    
1298
    for node in node_instance:
1299
      if (not node == node_current):
1300
        test = instance in node_instance[node]
1301
        _ErrorIf(test, self.EINSTANCEWRONGNODE, instance,
1302
                 "instance should not run on node %s", node)
1303

    
1304
  def _VerifyOrphanVolumes(self, node_vol_should, node_vol_is):
1305
    """Verify if there are any unknown volumes in the cluster.
1306

1307
    The .os, .swap and backup volumes are ignored. All other volumes are
1308
    reported as unknown.
1309

1310
    """
1311
    for node in node_vol_is:
1312
      for volume in node_vol_is[node]:
1313
        test = (node not in node_vol_should or
1314
                volume not in node_vol_should[node])
1315
        self._ErrorIf(test, self.ENODEORPHANLV, node,
1316
                      "volume %s is unknown", volume)
1317

    
1318
  def _VerifyOrphanInstances(self, instancelist, node_instance):
1319
    """Verify the list of running instances.
1320

1321
    This checks what instances are running but unknown to the cluster.
1322

1323
    """
1324
    for node in node_instance:
1325
      for o_inst in node_instance[node]:
1326
        test = o_inst not in instancelist
1327
        self._ErrorIf(test, self.ENODEORPHANINSTANCE, node,
1328
                      "instance %s on node %s should not exist", o_inst, node)
1329

    
1330
  def _VerifyNPlusOneMemory(self, node_info, instance_cfg):
1331
    """Verify N+1 Memory Resilience.
1332

1333
    Check that if one single node dies we can still start all the instances it
1334
    was primary for.
1335

1336
    """
1337
    for node, nodeinfo in node_info.iteritems():
1338
      # This code checks that every node which is now listed as secondary has
1339
      # enough memory to host all instances it is supposed to should a single
1340
      # other node in the cluster fail.
1341
      # FIXME: not ready for failover to an arbitrary node
1342
      # FIXME: does not support file-backed instances
1343
      # WARNING: we currently take into account down instances as well as up
1344
      # ones, considering that even if they're down someone might want to start
1345
      # them even in the event of a node failure.
1346
      for prinode, instances in nodeinfo['sinst-by-pnode'].iteritems():
1347
        needed_mem = 0
1348
        for instance in instances:
1349
          bep = self.cfg.GetClusterInfo().FillBE(instance_cfg[instance])
1350
          if bep[constants.BE_AUTO_BALANCE]:
1351
            needed_mem += bep[constants.BE_MEMORY]
1352
        test = nodeinfo['mfree'] < needed_mem
1353
        self._ErrorIf(test, self.ENODEN1, node,
1354
                      "not enough memory on to accommodate"
1355
                      " failovers should peer node %s fail", prinode)
1356

    
1357
  def CheckPrereq(self):
1358
    """Check prerequisites.
1359

1360
    Transform the list of checks we're going to skip into a set and check that
1361
    all its members are valid.
1362

1363
    """
1364
    self.skip_set = frozenset(self.op.skip_checks)
1365
    if not constants.VERIFY_OPTIONAL_CHECKS.issuperset(self.skip_set):
1366
      raise errors.OpPrereqError("Invalid checks to be skipped specified",
1367
                                 errors.ECODE_INVAL)
1368

    
1369
  def BuildHooksEnv(self):
1370
    """Build hooks env.
1371

1372
    Cluster-Verify hooks just ran in the post phase and their failure makes
1373
    the output be logged in the verify output and the verification to fail.
1374

1375
    """
1376
    all_nodes = self.cfg.GetNodeList()
1377
    env = {
1378
      "CLUSTER_TAGS": " ".join(self.cfg.GetClusterInfo().GetTags())
1379
      }
1380
    for node in self.cfg.GetAllNodesInfo().values():
1381
      env["NODE_TAGS_%s" % node.name] = " ".join(node.GetTags())
1382

    
1383
    return env, [], all_nodes
1384

    
1385
  def Exec(self, feedback_fn):
1386
    """Verify integrity of cluster, performing various test on nodes.
1387

1388
    """
1389
    self.bad = False
1390
    _ErrorIf = self._ErrorIf # pylint: disable-msg=C0103
1391
    verbose = self.op.verbose
1392
    self._feedback_fn = feedback_fn
1393
    feedback_fn("* Verifying global settings")
1394
    for msg in self.cfg.VerifyConfig():
1395
      _ErrorIf(True, self.ECLUSTERCFG, None, msg)
1396

    
1397
    # Check the cluster certificates
1398
    for cert_filename in constants.ALL_CERT_FILES:
1399
      (errcode, msg) = _VerifyCertificate(cert_filename)
1400
      _ErrorIf(errcode, self.ECLUSTERCERT, None, msg, code=errcode)
1401

    
1402
    vg_name = self.cfg.GetVGName()
1403
    hypervisors = self.cfg.GetClusterInfo().enabled_hypervisors
1404
    nodelist = utils.NiceSort(self.cfg.GetNodeList())
1405
    nodeinfo = [self.cfg.GetNodeInfo(nname) for nname in nodelist]
1406
    instancelist = utils.NiceSort(self.cfg.GetInstanceList())
1407
    instanceinfo = dict((iname, self.cfg.GetInstanceInfo(iname))
1408
                        for iname in instancelist)
1409
    i_non_redundant = [] # Non redundant instances
1410
    i_non_a_balanced = [] # Non auto-balanced instances
1411
    n_offline = [] # List of offline nodes
1412
    n_drained = [] # List of nodes being drained
1413
    node_volume = {}
1414
    node_instance = {}
1415
    node_info = {}
1416
    instance_cfg = {}
1417

    
1418
    # FIXME: verify OS list
1419
    # do local checksums
1420
    master_files = [constants.CLUSTER_CONF_FILE]
1421

    
1422
    file_names = ssconf.SimpleStore().GetFileList()
1423
    file_names.extend(constants.ALL_CERT_FILES)
1424
    file_names.extend(master_files)
1425

    
1426
    local_checksums = utils.FingerprintFiles(file_names)
1427

    
1428
    feedback_fn("* Gathering data (%d nodes)" % len(nodelist))
1429
    node_verify_param = {
1430
      constants.NV_FILELIST: file_names,
1431
      constants.NV_NODELIST: [node.name for node in nodeinfo
1432
                              if not node.offline],
1433
      constants.NV_HYPERVISOR: hypervisors,
1434
      constants.NV_NODENETTEST: [(node.name, node.primary_ip,
1435
                                  node.secondary_ip) for node in nodeinfo
1436
                                 if not node.offline],
1437
      constants.NV_INSTANCELIST: hypervisors,
1438
      constants.NV_VERSION: None,
1439
      constants.NV_HVINFO: self.cfg.GetHypervisorType(),
1440
      constants.NV_NODESETUP: None,
1441
      constants.NV_TIME: None,
1442
      }
1443

    
1444
    if vg_name is not None:
1445
      node_verify_param[constants.NV_VGLIST] = None
1446
      node_verify_param[constants.NV_LVLIST] = vg_name
1447
      node_verify_param[constants.NV_PVLIST] = [vg_name]
1448
      node_verify_param[constants.NV_DRBDLIST] = None
1449

    
1450
    # Due to the way our RPC system works, exact response times cannot be
1451
    # guaranteed (e.g. a broken node could run into a timeout). By keeping the
1452
    # time before and after executing the request, we can at least have a time
1453
    # window.
1454
    nvinfo_starttime = time.time()
1455
    all_nvinfo = self.rpc.call_node_verify(nodelist, node_verify_param,
1456
                                           self.cfg.GetClusterName())
1457
    nvinfo_endtime = time.time()
1458

    
1459
    cluster = self.cfg.GetClusterInfo()
1460
    master_node = self.cfg.GetMasterNode()
1461
    all_drbd_map = self.cfg.ComputeDRBDMap()
1462

    
1463
    feedback_fn("* Verifying node status")
1464
    for node_i in nodeinfo:
1465
      node = node_i.name
1466

    
1467
      if node_i.offline:
1468
        if verbose:
1469
          feedback_fn("* Skipping offline node %s" % (node,))
1470
        n_offline.append(node)
1471
        continue
1472

    
1473
      if node == master_node:
1474
        ntype = "master"
1475
      elif node_i.master_candidate:
1476
        ntype = "master candidate"
1477
      elif node_i.drained:
1478
        ntype = "drained"
1479
        n_drained.append(node)
1480
      else:
1481
        ntype = "regular"
1482
      if verbose:
1483
        feedback_fn("* Verifying node %s (%s)" % (node, ntype))
1484

    
1485
      msg = all_nvinfo[node].fail_msg
1486
      _ErrorIf(msg, self.ENODERPC, node, "while contacting node: %s", msg)
1487
      if msg:
1488
        continue
1489

    
1490
      nresult = all_nvinfo[node].payload
1491
      node_drbd = {}
1492
      for minor, instance in all_drbd_map[node].items():
1493
        test = instance not in instanceinfo
1494
        _ErrorIf(test, self.ECLUSTERCFG, None,
1495
                 "ghost instance '%s' in temporary DRBD map", instance)
1496
          # ghost instance should not be running, but otherwise we
1497
          # don't give double warnings (both ghost instance and
1498
          # unallocated minor in use)
1499
        if test:
1500
          node_drbd[minor] = (instance, False)
1501
        else:
1502
          instance = instanceinfo[instance]
1503
          node_drbd[minor] = (instance.name, instance.admin_up)
1504

    
1505
      self._VerifyNode(node_i, file_names, local_checksums,
1506
                       nresult, master_files, node_drbd, vg_name)
1507

    
1508
      lvdata = nresult.get(constants.NV_LVLIST, "Missing LV data")
1509
      if vg_name is None:
1510
        node_volume[node] = {}
1511
      elif isinstance(lvdata, basestring):
1512
        _ErrorIf(True, self.ENODELVM, node, "LVM problem on node: %s",
1513
                 utils.SafeEncode(lvdata))
1514
        node_volume[node] = {}
1515
      elif not isinstance(lvdata, dict):
1516
        _ErrorIf(True, self.ENODELVM, node, "rpc call to node failed (lvlist)")
1517
        continue
1518
      else:
1519
        node_volume[node] = lvdata
1520

    
1521
      # node_instance
1522
      idata = nresult.get(constants.NV_INSTANCELIST, None)
1523
      test = not isinstance(idata, list)
1524
      _ErrorIf(test, self.ENODEHV, node,
1525
               "rpc call to node failed (instancelist): %s",
1526
               utils.SafeEncode(str(idata)))
1527
      if test:
1528
        continue
1529

    
1530
      node_instance[node] = idata
1531

    
1532
      # node_info
1533
      nodeinfo = nresult.get(constants.NV_HVINFO, None)
1534
      test = not isinstance(nodeinfo, dict)
1535
      _ErrorIf(test, self.ENODEHV, node, "rpc call to node failed (hvinfo)")
1536
      if test:
1537
        continue
1538

    
1539
      # Node time
1540
      ntime = nresult.get(constants.NV_TIME, None)
1541
      try:
1542
        ntime_merged = utils.MergeTime(ntime)
1543
      except (ValueError, TypeError):
1544
        _ErrorIf(True, self.ENODETIME, node, "Node returned invalid time")
1545

    
1546
      if ntime_merged < (nvinfo_starttime - constants.NODE_MAX_CLOCK_SKEW):
1547
        ntime_diff = "%.01fs" % abs(nvinfo_starttime - ntime_merged)
1548
      elif ntime_merged > (nvinfo_endtime + constants.NODE_MAX_CLOCK_SKEW):
1549
        ntime_diff = "%.01fs" % abs(ntime_merged - nvinfo_endtime)
1550
      else:
1551
        ntime_diff = None
1552

    
1553
      _ErrorIf(ntime_diff is not None, self.ENODETIME, node,
1554
               "Node time diverges by at least %s from master node time",
1555
               ntime_diff)
1556

    
1557
      if ntime_diff is not None:
1558
        continue
1559

    
1560
      try:
1561
        node_info[node] = {
1562
          "mfree": int(nodeinfo['memory_free']),
1563
          "pinst": [],
1564
          "sinst": [],
1565
          # dictionary holding all instances this node is secondary for,
1566
          # grouped by their primary node. Each key is a cluster node, and each
1567
          # value is a list of instances which have the key as primary and the
1568
          # current node as secondary.  this is handy to calculate N+1 memory
1569
          # availability if you can only failover from a primary to its
1570
          # secondary.
1571
          "sinst-by-pnode": {},
1572
        }
1573
        # FIXME: devise a free space model for file based instances as well
1574
        if vg_name is not None:
1575
          test = (constants.NV_VGLIST not in nresult or
1576
                  vg_name not in nresult[constants.NV_VGLIST])
1577
          _ErrorIf(test, self.ENODELVM, node,
1578
                   "node didn't return data for the volume group '%s'"
1579
                   " - it is either missing or broken", vg_name)
1580
          if test:
1581
            continue
1582
          node_info[node]["dfree"] = int(nresult[constants.NV_VGLIST][vg_name])
1583
      except (ValueError, KeyError):
1584
        _ErrorIf(True, self.ENODERPC, node,
1585
                 "node returned invalid nodeinfo, check lvm/hypervisor")
1586
        continue
1587

    
1588
    node_vol_should = {}
1589

    
1590
    feedback_fn("* Verifying instance status")
1591
    for instance in instancelist:
1592
      if verbose:
1593
        feedback_fn("* Verifying instance %s" % instance)
1594
      inst_config = instanceinfo[instance]
1595
      self._VerifyInstance(instance, inst_config, node_volume,
1596
                           node_instance, n_offline)
1597
      inst_nodes_offline = []
1598

    
1599
      inst_config.MapLVsByNode(node_vol_should)
1600

    
1601
      instance_cfg[instance] = inst_config
1602

    
1603
      pnode = inst_config.primary_node
1604
      _ErrorIf(pnode not in node_info and pnode not in n_offline,
1605
               self.ENODERPC, pnode, "instance %s, connection to"
1606
               " primary node failed", instance)
1607
      if pnode in node_info:
1608
        node_info[pnode]['pinst'].append(instance)
1609

    
1610
      if pnode in n_offline:
1611
        inst_nodes_offline.append(pnode)
1612

    
1613
      # If the instance is non-redundant we cannot survive losing its primary
1614
      # node, so we are not N+1 compliant. On the other hand we have no disk
1615
      # templates with more than one secondary so that situation is not well
1616
      # supported either.
1617
      # FIXME: does not support file-backed instances
1618
      if len(inst_config.secondary_nodes) == 0:
1619
        i_non_redundant.append(instance)
1620
      _ErrorIf(len(inst_config.secondary_nodes) > 1,
1621
               self.EINSTANCELAYOUT, instance,
1622
               "instance has multiple secondary nodes", code="WARNING")
1623

    
1624
      if not cluster.FillBE(inst_config)[constants.BE_AUTO_BALANCE]:
1625
        i_non_a_balanced.append(instance)
1626

    
1627
      for snode in inst_config.secondary_nodes:
1628
        _ErrorIf(snode not in node_info and snode not in n_offline,
1629
                 self.ENODERPC, snode,
1630
                 "instance %s, connection to secondary node"
1631
                 " failed", instance)
1632

    
1633
        if snode in node_info:
1634
          node_info[snode]['sinst'].append(instance)
1635
          if pnode not in node_info[snode]['sinst-by-pnode']:
1636
            node_info[snode]['sinst-by-pnode'][pnode] = []
1637
          node_info[snode]['sinst-by-pnode'][pnode].append(instance)
1638

    
1639
        if snode in n_offline:
1640
          inst_nodes_offline.append(snode)
1641

    
1642
      # warn that the instance lives on offline nodes
1643
      _ErrorIf(inst_nodes_offline, self.EINSTANCEBADNODE, instance,
1644
               "instance lives on offline node(s) %s",
1645
               utils.CommaJoin(inst_nodes_offline))
1646

    
1647
    feedback_fn("* Verifying orphan volumes")
1648
    self._VerifyOrphanVolumes(node_vol_should, node_volume)
1649

    
1650
    feedback_fn("* Verifying remaining instances")
1651
    self._VerifyOrphanInstances(instancelist, node_instance)
1652

    
1653
    if constants.VERIFY_NPLUSONE_MEM not in self.skip_set:
1654
      feedback_fn("* Verifying N+1 Memory redundancy")
1655
      self._VerifyNPlusOneMemory(node_info, instance_cfg)
1656

    
1657
    feedback_fn("* Other Notes")
1658
    if i_non_redundant:
1659
      feedback_fn("  - NOTICE: %d non-redundant instance(s) found."
1660
                  % len(i_non_redundant))
1661

    
1662
    if i_non_a_balanced:
1663
      feedback_fn("  - NOTICE: %d non-auto-balanced instance(s) found."
1664
                  % len(i_non_a_balanced))
1665

    
1666
    if n_offline:
1667
      feedback_fn("  - NOTICE: %d offline node(s) found." % len(n_offline))
1668

    
1669
    if n_drained:
1670
      feedback_fn("  - NOTICE: %d drained node(s) found." % len(n_drained))
1671

    
1672
    return not self.bad
1673

    
1674
  def HooksCallBack(self, phase, hooks_results, feedback_fn, lu_result):
1675
    """Analyze the post-hooks' result
1676

1677
    This method analyses the hook result, handles it, and sends some
1678
    nicely-formatted feedback back to the user.
1679

1680
    @param phase: one of L{constants.HOOKS_PHASE_POST} or
1681
        L{constants.HOOKS_PHASE_PRE}; it denotes the hooks phase
1682
    @param hooks_results: the results of the multi-node hooks rpc call
1683
    @param feedback_fn: function used send feedback back to the caller
1684
    @param lu_result: previous Exec result
1685
    @return: the new Exec result, based on the previous result
1686
        and hook results
1687

1688
    """
1689
    # We only really run POST phase hooks, and are only interested in
1690
    # their results
1691
    if phase == constants.HOOKS_PHASE_POST:
1692
      # Used to change hooks' output to proper indentation
1693
      indent_re = re.compile('^', re.M)
1694
      feedback_fn("* Hooks Results")
1695
      assert hooks_results, "invalid result from hooks"
1696

    
1697
      for node_name in hooks_results:
1698
        res = hooks_results[node_name]
1699
        msg = res.fail_msg
1700
        test = msg and not res.offline
1701
        self._ErrorIf(test, self.ENODEHOOKS, node_name,
1702
                      "Communication failure in hooks execution: %s", msg)
1703
        if res.offline or msg:
1704
          # No need to investigate payload if node is offline or gave an error.
1705
          # override manually lu_result here as _ErrorIf only
1706
          # overrides self.bad
1707
          lu_result = 1
1708
          continue
1709
        for script, hkr, output in res.payload:
1710
          test = hkr == constants.HKR_FAIL
1711
          self._ErrorIf(test, self.ENODEHOOKS, node_name,
1712
                        "Script %s failed, output:", script)
1713
          if test:
1714
            output = indent_re.sub('      ', output)
1715
            feedback_fn("%s" % output)
1716
            lu_result = 0
1717

    
1718
      return lu_result
1719

    
1720

    
1721
class LUVerifyDisks(NoHooksLU):
1722
  """Verifies the cluster disks status.
1723

1724
  """
1725
  _OP_REQP = []
1726
  REQ_BGL = False
1727

    
1728
  def ExpandNames(self):
1729
    self.needed_locks = {
1730
      locking.LEVEL_NODE: locking.ALL_SET,
1731
      locking.LEVEL_INSTANCE: locking.ALL_SET,
1732
    }
1733
    self.share_locks = dict.fromkeys(locking.LEVELS, 1)
1734

    
1735
  def CheckPrereq(self):
1736
    """Check prerequisites.
1737

1738
    This has no prerequisites.
1739

1740
    """
1741
    pass
1742

    
1743
  def Exec(self, feedback_fn):
1744
    """Verify integrity of cluster disks.
1745

1746
    @rtype: tuple of three items
1747
    @return: a tuple of (dict of node-to-node_error, list of instances
1748
        which need activate-disks, dict of instance: (node, volume) for
1749
        missing volumes
1750

1751
    """
1752
    result = res_nodes, res_instances, res_missing = {}, [], {}
1753

    
1754
    vg_name = self.cfg.GetVGName()
1755
    nodes = utils.NiceSort(self.cfg.GetNodeList())
1756
    instances = [self.cfg.GetInstanceInfo(name)
1757
                 for name in self.cfg.GetInstanceList()]
1758

    
1759
    nv_dict = {}
1760
    for inst in instances:
1761
      inst_lvs = {}
1762
      if (not inst.admin_up or
1763
          inst.disk_template not in constants.DTS_NET_MIRROR):
1764
        continue
1765
      inst.MapLVsByNode(inst_lvs)
1766
      # transform { iname: {node: [vol,],},} to {(node, vol): iname}
1767
      for node, vol_list in inst_lvs.iteritems():
1768
        for vol in vol_list:
1769
          nv_dict[(node, vol)] = inst
1770

    
1771
    if not nv_dict:
1772
      return result
1773

    
1774
    node_lvs = self.rpc.call_lv_list(nodes, vg_name)
1775

    
1776
    for node in nodes:
1777
      # node_volume
1778
      node_res = node_lvs[node]
1779
      if node_res.offline:
1780
        continue
1781
      msg = node_res.fail_msg
1782
      if msg:
1783
        logging.warning("Error enumerating LVs on node %s: %s", node, msg)
1784
        res_nodes[node] = msg
1785
        continue
1786

    
1787
      lvs = node_res.payload
1788
      for lv_name, (_, _, lv_online) in lvs.items():
1789
        inst = nv_dict.pop((node, lv_name), None)
1790
        if (not lv_online and inst is not None
1791
            and inst.name not in res_instances):
1792
          res_instances.append(inst.name)
1793

    
1794
    # any leftover items in nv_dict are missing LVs, let's arrange the
1795
    # data better
1796
    for key, inst in nv_dict.iteritems():
1797
      if inst.name not in res_missing:
1798
        res_missing[inst.name] = []
1799
      res_missing[inst.name].append(key)
1800

    
1801
    return result
1802

    
1803

    
1804
class LURepairDiskSizes(NoHooksLU):
1805
  """Verifies the cluster disks sizes.
1806

1807
  """
1808
  _OP_REQP = ["instances"]
1809
  REQ_BGL = False
1810

    
1811
  def ExpandNames(self):
1812
    if not isinstance(self.op.instances, list):
1813
      raise errors.OpPrereqError("Invalid argument type 'instances'",
1814
                                 errors.ECODE_INVAL)
1815

    
1816
    if self.op.instances:
1817
      self.wanted_names = []
1818
      for name in self.op.instances:
1819
        full_name = _ExpandInstanceName(self.cfg, name)
1820
        self.wanted_names.append(full_name)
1821
      self.needed_locks = {
1822
        locking.LEVEL_NODE: [],
1823
        locking.LEVEL_INSTANCE: self.wanted_names,
1824
        }
1825
      self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
1826
    else:
1827
      self.wanted_names = None
1828
      self.needed_locks = {
1829
        locking.LEVEL_NODE: locking.ALL_SET,
1830
        locking.LEVEL_INSTANCE: locking.ALL_SET,
1831
        }
1832
    self.share_locks = dict(((i, 1) for i in locking.LEVELS))
1833

    
1834
  def DeclareLocks(self, level):
1835
    if level == locking.LEVEL_NODE and self.wanted_names is not None:
1836
      self._LockInstancesNodes(primary_only=True)
1837

    
1838
  def CheckPrereq(self):
1839
    """Check prerequisites.
1840

1841
    This only checks the optional instance list against the existing names.
1842

1843
    """
1844
    if self.wanted_names is None:
1845
      self.wanted_names = self.acquired_locks[locking.LEVEL_INSTANCE]
1846

    
1847
    self.wanted_instances = [self.cfg.GetInstanceInfo(name) for name
1848
                             in self.wanted_names]
1849

    
1850
  def _EnsureChildSizes(self, disk):
1851
    """Ensure children of the disk have the needed disk size.
1852

1853
    This is valid mainly for DRBD8 and fixes an issue where the
1854
    children have smaller disk size.
1855

1856
    @param disk: an L{ganeti.objects.Disk} object
1857

1858
    """
1859
    if disk.dev_type == constants.LD_DRBD8:
1860
      assert disk.children, "Empty children for DRBD8?"
1861
      fchild = disk.children[0]
1862
      mismatch = fchild.size < disk.size
1863
      if mismatch:
1864
        self.LogInfo("Child disk has size %d, parent %d, fixing",
1865
                     fchild.size, disk.size)
1866
        fchild.size = disk.size
1867

    
1868
      # and we recurse on this child only, not on the metadev
1869
      return self._EnsureChildSizes(fchild) or mismatch
1870
    else:
1871
      return False
1872

    
1873
  def Exec(self, feedback_fn):
1874
    """Verify the size of cluster disks.
1875

1876
    """
1877
    # TODO: check child disks too
1878
    # TODO: check differences in size between primary/secondary nodes
1879
    per_node_disks = {}
1880
    for instance in self.wanted_instances:
1881
      pnode = instance.primary_node
1882
      if pnode not in per_node_disks:
1883
        per_node_disks[pnode] = []
1884
      for idx, disk in enumerate(instance.disks):
1885
        per_node_disks[pnode].append((instance, idx, disk))
1886

    
1887
    changed = []
1888
    for node, dskl in per_node_disks.items():
1889
      newl = [v[2].Copy() for v in dskl]
1890
      for dsk in newl:
1891
        self.cfg.SetDiskID(dsk, node)
1892
      result = self.rpc.call_blockdev_getsizes(node, newl)
1893
      if result.fail_msg:
1894
        self.LogWarning("Failure in blockdev_getsizes call to node"
1895
                        " %s, ignoring", node)
1896
        continue
1897
      if len(result.data) != len(dskl):
1898
        self.LogWarning("Invalid result from node %s, ignoring node results",
1899
                        node)
1900
        continue
1901
      for ((instance, idx, disk), size) in zip(dskl, result.data):
1902
        if size is None:
1903
          self.LogWarning("Disk %d of instance %s did not return size"
1904
                          " information, ignoring", idx, instance.name)
1905
          continue
1906
        if not isinstance(size, (int, long)):
1907
          self.LogWarning("Disk %d of instance %s did not return valid"
1908
                          " size information, ignoring", idx, instance.name)
1909
          continue
1910
        size = size >> 20
1911
        if size != disk.size:
1912
          self.LogInfo("Disk %d of instance %s has mismatched size,"
1913
                       " correcting: recorded %d, actual %d", idx,
1914
                       instance.name, disk.size, size)
1915
          disk.size = size
1916
          self.cfg.Update(instance, feedback_fn)
1917
          changed.append((instance.name, idx, size))
1918
        if self._EnsureChildSizes(disk):
1919
          self.cfg.Update(instance, feedback_fn)
1920
          changed.append((instance.name, idx, disk.size))
1921
    return changed
1922

    
1923

    
1924
class LURenameCluster(LogicalUnit):
1925
  """Rename the cluster.
1926

1927
  """
1928
  HPATH = "cluster-rename"
1929
  HTYPE = constants.HTYPE_CLUSTER
1930
  _OP_REQP = ["name"]
1931

    
1932
  def BuildHooksEnv(self):
1933
    """Build hooks env.
1934

1935
    """
1936
    env = {
1937
      "OP_TARGET": self.cfg.GetClusterName(),
1938
      "NEW_NAME": self.op.name,
1939
      }
1940
    mn = self.cfg.GetMasterNode()
1941
    all_nodes = self.cfg.GetNodeList()
1942
    return env, [mn], all_nodes
1943

    
1944
  def CheckPrereq(self):
1945
    """Verify that the passed name is a valid one.
1946

1947
    """
1948
    hostname = utils.GetHostInfo(self.op.name)
1949

    
1950
    new_name = hostname.name
1951
    self.ip = new_ip = hostname.ip
1952
    old_name = self.cfg.GetClusterName()
1953
    old_ip = self.cfg.GetMasterIP()
1954
    if new_name == old_name and new_ip == old_ip:
1955
      raise errors.OpPrereqError("Neither the name nor the IP address of the"
1956
                                 " cluster has changed",
1957
                                 errors.ECODE_INVAL)
1958
    if new_ip != old_ip:
1959
      if utils.TcpPing(new_ip, constants.DEFAULT_NODED_PORT):
1960
        raise errors.OpPrereqError("The given cluster IP address (%s) is"
1961
                                   " reachable on the network. Aborting." %
1962
                                   new_ip, errors.ECODE_NOTUNIQUE)
1963

    
1964
    self.op.name = new_name
1965

    
1966
  def Exec(self, feedback_fn):
1967
    """Rename the cluster.
1968

1969
    """
1970
    clustername = self.op.name
1971
    ip = self.ip
1972

    
1973
    # shutdown the master IP
1974
    master = self.cfg.GetMasterNode()
1975
    result = self.rpc.call_node_stop_master(master, False)
1976
    result.Raise("Could not disable the master role")
1977

    
1978
    try:
1979
      cluster = self.cfg.GetClusterInfo()
1980
      cluster.cluster_name = clustername
1981
      cluster.master_ip = ip
1982
      self.cfg.Update(cluster, feedback_fn)
1983

    
1984
      # update the known hosts file
1985
      ssh.WriteKnownHostsFile(self.cfg, constants.SSH_KNOWN_HOSTS_FILE)
1986
      node_list = self.cfg.GetNodeList()
1987
      try:
1988
        node_list.remove(master)
1989
      except ValueError:
1990
        pass
1991
      result = self.rpc.call_upload_file(node_list,
1992
                                         constants.SSH_KNOWN_HOSTS_FILE)
1993
      for to_node, to_result in result.iteritems():
1994
        msg = to_result.fail_msg
1995
        if msg:
1996
          msg = ("Copy of file %s to node %s failed: %s" %
1997
                 (constants.SSH_KNOWN_HOSTS_FILE, to_node, msg))
1998
          self.proc.LogWarning(msg)
1999

    
2000
    finally:
2001
      result = self.rpc.call_node_start_master(master, False, False)
2002
      msg = result.fail_msg
2003
      if msg:
2004
        self.LogWarning("Could not re-enable the master role on"
2005
                        " the master, please restart manually: %s", msg)
2006

    
2007

    
2008
def _RecursiveCheckIfLVMBased(disk):
2009
  """Check if the given disk or its children are lvm-based.
2010

2011
  @type disk: L{objects.Disk}
2012
  @param disk: the disk to check
2013
  @rtype: boolean
2014
  @return: boolean indicating whether a LD_LV dev_type was found or not
2015

2016
  """
2017
  if disk.children:
2018
    for chdisk in disk.children:
2019
      if _RecursiveCheckIfLVMBased(chdisk):
2020
        return True
2021
  return disk.dev_type == constants.LD_LV
2022

    
2023

    
2024
class LUSetClusterParams(LogicalUnit):
2025
  """Change the parameters of the cluster.
2026

2027
  """
2028
  HPATH = "cluster-modify"
2029
  HTYPE = constants.HTYPE_CLUSTER
2030
  _OP_REQP = []
2031
  REQ_BGL = False
2032

    
2033
  def CheckArguments(self):
2034
    """Check parameters
2035

2036
    """
2037
    if not hasattr(self.op, "candidate_pool_size"):
2038
      self.op.candidate_pool_size = None
2039
    if self.op.candidate_pool_size is not None:
2040
      try:
2041
        self.op.candidate_pool_size = int(self.op.candidate_pool_size)
2042
      except (ValueError, TypeError), err:
2043
        raise errors.OpPrereqError("Invalid candidate_pool_size value: %s" %
2044
                                   str(err), errors.ECODE_INVAL)
2045
      if self.op.candidate_pool_size < 1:
2046
        raise errors.OpPrereqError("At least one master candidate needed",
2047
                                   errors.ECODE_INVAL)
2048

    
2049
  def ExpandNames(self):
2050
    # FIXME: in the future maybe other cluster params won't require checking on
2051
    # all nodes to be modified.
2052
    self.needed_locks = {
2053
      locking.LEVEL_NODE: locking.ALL_SET,
2054
    }
2055
    self.share_locks[locking.LEVEL_NODE] = 1
2056

    
2057
  def BuildHooksEnv(self):
2058
    """Build hooks env.
2059

2060
    """
2061
    env = {
2062
      "OP_TARGET": self.cfg.GetClusterName(),
2063
      "NEW_VG_NAME": self.op.vg_name,
2064
      }
2065
    mn = self.cfg.GetMasterNode()
2066
    return env, [mn], [mn]
2067

    
2068
  def CheckPrereq(self):
2069
    """Check prerequisites.
2070

2071
    This checks whether the given params don't conflict and
2072
    if the given volume group is valid.
2073

2074
    """
2075
    if self.op.vg_name is not None and not self.op.vg_name:
2076
      instances = self.cfg.GetAllInstancesInfo().values()
2077
      for inst in instances:
2078
        for disk in inst.disks:
2079
          if _RecursiveCheckIfLVMBased(disk):
2080
            raise errors.OpPrereqError("Cannot disable lvm storage while"
2081
                                       " lvm-based instances exist",
2082
                                       errors.ECODE_INVAL)
2083

    
2084
    node_list = self.acquired_locks[locking.LEVEL_NODE]
2085

    
2086
    # if vg_name not None, checks given volume group on all nodes
2087
    if self.op.vg_name:
2088
      vglist = self.rpc.call_vg_list(node_list)
2089
      for node in node_list:
2090
        msg = vglist[node].fail_msg
2091
        if msg:
2092
          # ignoring down node
2093
          self.LogWarning("Error while gathering data on node %s"
2094
                          " (ignoring node): %s", node, msg)
2095
          continue
2096
        vgstatus = utils.CheckVolumeGroupSize(vglist[node].payload,
2097
                                              self.op.vg_name,
2098
                                              constants.MIN_VG_SIZE)
2099
        if vgstatus:
2100
          raise errors.OpPrereqError("Error on node '%s': %s" %
2101
                                     (node, vgstatus), errors.ECODE_ENVIRON)
2102

    
2103
    self.cluster = cluster = self.cfg.GetClusterInfo()
2104
    # validate params changes
2105
    if self.op.beparams:
2106
      utils.ForceDictType(self.op.beparams, constants.BES_PARAMETER_TYPES)
2107
      self.new_beparams = objects.FillDict(
2108
        cluster.beparams[constants.PP_DEFAULT], self.op.beparams)
2109

    
2110
    if self.op.nicparams:
2111
      utils.ForceDictType(self.op.nicparams, constants.NICS_PARAMETER_TYPES)
2112
      self.new_nicparams = objects.FillDict(
2113
        cluster.nicparams[constants.PP_DEFAULT], self.op.nicparams)
2114
      objects.NIC.CheckParameterSyntax(self.new_nicparams)
2115
      nic_errors = []
2116

    
2117
      # check all instances for consistency
2118
      for instance in self.cfg.GetAllInstancesInfo().values():
2119
        for nic_idx, nic in enumerate(instance.nics):
2120
          params_copy = copy.deepcopy(nic.nicparams)
2121
          params_filled = objects.FillDict(self.new_nicparams, params_copy)
2122

    
2123
          # check parameter syntax
2124
          try:
2125
            objects.NIC.CheckParameterSyntax(params_filled)
2126
          except errors.ConfigurationError, err:
2127
            nic_errors.append("Instance %s, nic/%d: %s" %
2128
                              (instance.name, nic_idx, err))
2129

    
2130
          # if we're moving instances to routed, check that they have an ip
2131
          target_mode = params_filled[constants.NIC_MODE]
2132
          if target_mode == constants.NIC_MODE_ROUTED and not nic.ip:
2133
            nic_errors.append("Instance %s, nic/%d: routed nick with no ip" %
2134
                              (instance.name, nic_idx))
2135
      if nic_errors:
2136
        raise errors.OpPrereqError("Cannot apply the change, errors:\n%s" %
2137
                                   "\n".join(nic_errors))
2138

    
2139
    # hypervisor list/parameters
2140
    self.new_hvparams = objects.FillDict(cluster.hvparams, {})
2141
    if self.op.hvparams:
2142
      if not isinstance(self.op.hvparams, dict):
2143
        raise errors.OpPrereqError("Invalid 'hvparams' parameter on input",
2144
                                   errors.ECODE_INVAL)
2145
      for hv_name, hv_dict in self.op.hvparams.items():
2146
        if hv_name not in self.new_hvparams:
2147
          self.new_hvparams[hv_name] = hv_dict
2148
        else:
2149
          self.new_hvparams[hv_name].update(hv_dict)
2150

    
2151
    # os hypervisor parameters
2152
    self.new_os_hvp = objects.FillDict(cluster.os_hvp, {})
2153
    if self.op.os_hvp:
2154
      if not isinstance(self.op.os_hvp, dict):
2155
        raise errors.OpPrereqError("Invalid 'os_hvp' parameter on input",
2156
                                   errors.ECODE_INVAL)
2157
      for os_name, hvs in self.op.os_hvp.items():
2158
        if not isinstance(hvs, dict):
2159
          raise errors.OpPrereqError(("Invalid 'os_hvp' parameter on"
2160
                                      " input"), errors.ECODE_INVAL)
2161
        if os_name not in self.new_os_hvp:
2162
          self.new_os_hvp[os_name] = hvs
2163
        else:
2164
          for hv_name, hv_dict in hvs.items():
2165
            if hv_name not in self.new_os_hvp[os_name]:
2166
              self.new_os_hvp[os_name][hv_name] = hv_dict
2167
            else:
2168
              self.new_os_hvp[os_name][hv_name].update(hv_dict)
2169

    
2170
    if self.op.enabled_hypervisors is not None:
2171
      self.hv_list = self.op.enabled_hypervisors
2172
      if not self.hv_list:
2173
        raise errors.OpPrereqError("Enabled hypervisors list must contain at"
2174
                                   " least one member",
2175
                                   errors.ECODE_INVAL)
2176
      invalid_hvs = set(self.hv_list) - constants.HYPER_TYPES
2177
      if invalid_hvs:
2178
        raise errors.OpPrereqError("Enabled hypervisors contains invalid"
2179
                                   " entries: %s" %
2180
                                   utils.CommaJoin(invalid_hvs),
2181
                                   errors.ECODE_INVAL)
2182
    else:
2183
      self.hv_list = cluster.enabled_hypervisors
2184

    
2185
    if self.op.hvparams or self.op.enabled_hypervisors is not None:
2186
      # either the enabled list has changed, or the parameters have, validate
2187
      for hv_name, hv_params in self.new_hvparams.items():
2188
        if ((self.op.hvparams and hv_name in self.op.hvparams) or
2189
            (self.op.enabled_hypervisors and
2190
             hv_name in self.op.enabled_hypervisors)):
2191
          # either this is a new hypervisor, or its parameters have changed
2192
          hv_class = hypervisor.GetHypervisor(hv_name)
2193
          utils.ForceDictType(hv_params, constants.HVS_PARAMETER_TYPES)
2194
          hv_class.CheckParameterSyntax(hv_params)
2195
          _CheckHVParams(self, node_list, hv_name, hv_params)
2196

    
2197
    if self.op.os_hvp:
2198
      # no need to check any newly-enabled hypervisors, since the
2199
      # defaults have already been checked in the above code-block
2200
      for os_name, os_hvp in self.new_os_hvp.items():
2201
        for hv_name, hv_params in os_hvp.items():
2202
          utils.ForceDictType(hv_params, constants.HVS_PARAMETER_TYPES)
2203
          # we need to fill in the new os_hvp on top of the actual hv_p
2204
          cluster_defaults = self.new_hvparams.get(hv_name, {})
2205
          new_osp = objects.FillDict(cluster_defaults, hv_params)
2206
          hv_class = hypervisor.GetHypervisor(hv_name)
2207
          hv_class.CheckParameterSyntax(new_osp)
2208
          _CheckHVParams(self, node_list, hv_name, new_osp)
2209

    
2210

    
2211
  def Exec(self, feedback_fn):
2212
    """Change the parameters of the cluster.
2213

2214
    """
2215
    if self.op.vg_name is not None:
2216
      new_volume = self.op.vg_name
2217
      if not new_volume:
2218
        new_volume = None
2219
      if new_volume != self.cfg.GetVGName():
2220
        self.cfg.SetVGName(new_volume)
2221
      else:
2222
        feedback_fn("Cluster LVM configuration already in desired"
2223
                    " state, not changing")
2224
    if self.op.hvparams:
2225
      self.cluster.hvparams = self.new_hvparams
2226
    if self.op.os_hvp:
2227
      self.cluster.os_hvp = self.new_os_hvp
2228
    if self.op.enabled_hypervisors is not None:
2229
      self.cluster.enabled_hypervisors = self.op.enabled_hypervisors
2230
    if self.op.beparams:
2231
      self.cluster.beparams[constants.PP_DEFAULT] = self.new_beparams
2232
    if self.op.nicparams:
2233
      self.cluster.nicparams[constants.PP_DEFAULT] = self.new_nicparams
2234

    
2235
    if self.op.candidate_pool_size is not None:
2236
      self.cluster.candidate_pool_size = self.op.candidate_pool_size
2237
      # we need to update the pool size here, otherwise the save will fail
2238
      _AdjustCandidatePool(self, [])
2239

    
2240
    self.cfg.Update(self.cluster, feedback_fn)
2241

    
2242

    
2243
def _RedistributeAncillaryFiles(lu, additional_nodes=None):
2244
  """Distribute additional files which are part of the cluster configuration.
2245

2246
  ConfigWriter takes care of distributing the config and ssconf files, but
2247
  there are more files which should be distributed to all nodes. This function
2248
  makes sure those are copied.
2249

2250
  @param lu: calling logical unit
2251
  @param additional_nodes: list of nodes not in the config to distribute to
2252

2253
  """
2254
  # 1. Gather target nodes
2255
  myself = lu.cfg.GetNodeInfo(lu.cfg.GetMasterNode())
2256
  dist_nodes = lu.cfg.GetOnlineNodeList()
2257
  if additional_nodes is not None:
2258
    dist_nodes.extend(additional_nodes)
2259
  if myself.name in dist_nodes:
2260
    dist_nodes.remove(myself.name)
2261

    
2262
  # 2. Gather files to distribute
2263
  dist_files = set([constants.ETC_HOSTS,
2264
                    constants.SSH_KNOWN_HOSTS_FILE,
2265
                    constants.RAPI_CERT_FILE,
2266
                    constants.RAPI_USERS_FILE,
2267
                    constants.HMAC_CLUSTER_KEY,
2268
                   ])
2269

    
2270
  enabled_hypervisors = lu.cfg.GetClusterInfo().enabled_hypervisors
2271
  for hv_name in enabled_hypervisors:
2272
    hv_class = hypervisor.GetHypervisor(hv_name)
2273
    dist_files.update(hv_class.GetAncillaryFiles())
2274

    
2275
  # 3. Perform the files upload
2276
  for fname in dist_files:
2277
    if os.path.exists(fname):
2278
      result = lu.rpc.call_upload_file(dist_nodes, fname)
2279
      for to_node, to_result in result.items():
2280
        msg = to_result.fail_msg
2281
        if msg:
2282
          msg = ("Copy of file %s to node %s failed: %s" %
2283
                 (fname, to_node, msg))
2284
          lu.proc.LogWarning(msg)
2285

    
2286

    
2287
class LURedistributeConfig(NoHooksLU):
2288
  """Force the redistribution of cluster configuration.
2289

2290
  This is a very simple LU.
2291

2292
  """
2293
  _OP_REQP = []
2294
  REQ_BGL = False
2295

    
2296
  def ExpandNames(self):
2297
    self.needed_locks = {
2298
      locking.LEVEL_NODE: locking.ALL_SET,
2299
    }
2300
    self.share_locks[locking.LEVEL_NODE] = 1
2301

    
2302
  def CheckPrereq(self):
2303
    """Check prerequisites.
2304

2305
    """
2306

    
2307
  def Exec(self, feedback_fn):
2308
    """Redistribute the configuration.
2309

2310
    """
2311
    self.cfg.Update(self.cfg.GetClusterInfo(), feedback_fn)
2312
    _RedistributeAncillaryFiles(self)
2313

    
2314

    
2315
def _WaitForSync(lu, instance, oneshot=False):
2316
  """Sleep and poll for an instance's disk to sync.
2317

2318
  """
2319
  if not instance.disks:
2320
    return True
2321

    
2322
  if not oneshot:
2323
    lu.proc.LogInfo("Waiting for instance %s to sync disks." % instance.name)
2324

    
2325
  node = instance.primary_node
2326

    
2327
  for dev in instance.disks:
2328
    lu.cfg.SetDiskID(dev, node)
2329

    
2330
  # TODO: Convert to utils.Retry
2331

    
2332
  retries = 0
2333
  degr_retries = 10 # in seconds, as we sleep 1 second each time
2334
  while True:
2335
    max_time = 0
2336
    done = True
2337
    cumul_degraded = False
2338
    rstats = lu.rpc.call_blockdev_getmirrorstatus(node, instance.disks)
2339
    msg = rstats.fail_msg
2340
    if msg:
2341
      lu.LogWarning("Can't get any data from node %s: %s", node, msg)
2342
      retries += 1
2343
      if retries >= 10:
2344
        raise errors.RemoteError("Can't contact node %s for mirror data,"
2345
                                 " aborting." % node)
2346
      time.sleep(6)
2347
      continue
2348
    rstats = rstats.payload
2349
    retries = 0
2350
    for i, mstat in enumerate(rstats):
2351
      if mstat is None:
2352
        lu.LogWarning("Can't compute data for node %s/%s",
2353
                           node, instance.disks[i].iv_name)
2354
        continue
2355

    
2356
      cumul_degraded = (cumul_degraded or
2357
                        (mstat.is_degraded and mstat.sync_percent is None))
2358
      if mstat.sync_percent is not None:
2359
        done = False
2360
        if mstat.estimated_time is not None:
2361
          rem_time = "%d estimated seconds remaining" % mstat.estimated_time
2362
          max_time = mstat.estimated_time
2363
        else:
2364
          rem_time = "no time estimate"
2365
        lu.proc.LogInfo("- device %s: %5.2f%% done, %s" %
2366
                        (instance.disks[i].iv_name, mstat.sync_percent,
2367
                         rem_time))
2368

    
2369
    # if we're done but degraded, let's do a few small retries, to
2370
    # make sure we see a stable and not transient situation; therefore
2371
    # we force restart of the loop
2372
    if (done or oneshot) and cumul_degraded and degr_retries > 0:
2373
      logging.info("Degraded disks found, %d retries left", degr_retries)
2374
      degr_retries -= 1
2375
      time.sleep(1)
2376
      continue
2377

    
2378
    if done or oneshot:
2379
      break
2380

    
2381
    time.sleep(min(60, max_time))
2382

    
2383
  if done:
2384
    lu.proc.LogInfo("Instance %s's disks are in sync." % instance.name)
2385
  return not cumul_degraded
2386

    
2387

    
2388
def _CheckDiskConsistency(lu, dev, node, on_primary, ldisk=False):
2389
  """Check that mirrors are not degraded.
2390

2391
  The ldisk parameter, if True, will change the test from the
2392
  is_degraded attribute (which represents overall non-ok status for
2393
  the device(s)) to the ldisk (representing the local storage status).
2394

2395
  """
2396
  lu.cfg.SetDiskID(dev, node)
2397

    
2398
  result = True
2399

    
2400
  if on_primary or dev.AssembleOnSecondary():
2401
    rstats = lu.rpc.call_blockdev_find(node, dev)
2402
    msg = rstats.fail_msg
2403
    if msg:
2404
      lu.LogWarning("Can't find disk on node %s: %s", node, msg)
2405
      result = False
2406
    elif not rstats.payload:
2407
      lu.LogWarning("Can't find disk on node %s", node)
2408
      result = False
2409
    else:
2410
      if ldisk:
2411
        result = result and rstats.payload.ldisk_status == constants.LDS_OKAY
2412
      else:
2413
        result = result and not rstats.payload.is_degraded
2414

    
2415
  if dev.children:
2416
    for child in dev.children:
2417
      result = result and _CheckDiskConsistency(lu, child, node, on_primary)
2418

    
2419
  return result
2420

    
2421

    
2422
class LUDiagnoseOS(NoHooksLU):
2423
  """Logical unit for OS diagnose/query.
2424

2425
  """
2426
  _OP_REQP = ["output_fields", "names"]
2427
  REQ_BGL = False
2428
  _FIELDS_STATIC = utils.FieldSet()
2429
  _FIELDS_DYNAMIC = utils.FieldSet("name", "valid", "node_status", "variants")
2430
  # Fields that need calculation of global os validity
2431
  _FIELDS_NEEDVALID = frozenset(["valid", "variants"])
2432

    
2433
  def ExpandNames(self):
2434
    if self.op.names:
2435
      raise errors.OpPrereqError("Selective OS query not supported",
2436
                                 errors.ECODE_INVAL)
2437

    
2438
    _CheckOutputFields(static=self._FIELDS_STATIC,
2439
                       dynamic=self._FIELDS_DYNAMIC,
2440
                       selected=self.op.output_fields)
2441

    
2442
    # Lock all nodes, in shared mode
2443
    # Temporary removal of locks, should be reverted later
2444
    # TODO: reintroduce locks when they are lighter-weight
2445
    self.needed_locks = {}
2446
    #self.share_locks[locking.LEVEL_NODE] = 1
2447
    #self.needed_locks[locking.LEVEL_NODE] = locking.ALL_SET
2448

    
2449
  def CheckPrereq(self):
2450
    """Check prerequisites.
2451

2452
    """
2453

    
2454
  @staticmethod
2455
  def _DiagnoseByOS(rlist):
2456
    """Remaps a per-node return list into an a per-os per-node dictionary
2457

2458
    @param rlist: a map with node names as keys and OS objects as values
2459

2460
    @rtype: dict
2461
    @return: a dictionary with osnames as keys and as value another map, with
2462
        nodes as keys and tuples of (path, status, diagnose) as values, eg::
2463

2464
          {"debian-etch": {"node1": [(/usr/lib/..., True, ""),
2465
                                     (/srv/..., False, "invalid api")],
2466
                           "node2": [(/srv/..., True, "")]}
2467
          }
2468

2469
    """
2470
    all_os = {}
2471
    # we build here the list of nodes that didn't fail the RPC (at RPC
2472
    # level), so that nodes with a non-responding node daemon don't
2473
    # make all OSes invalid
2474
    good_nodes = [node_name for node_name in rlist
2475
                  if not rlist[node_name].fail_msg]
2476
    for node_name, nr in rlist.items():
2477
      if nr.fail_msg or not nr.payload:
2478
        continue
2479
      for name, path, status, diagnose, variants in nr.payload:
2480
        if name not in all_os:
2481
          # build a list of nodes for this os containing empty lists
2482
          # for each node in node_list
2483
          all_os[name] = {}
2484
          for nname in good_nodes:
2485
            all_os[name][nname] = []
2486
        all_os[name][node_name].append((path, status, diagnose, variants))
2487
    return all_os
2488

    
2489
  def Exec(self, feedback_fn):
2490
    """Compute the list of OSes.
2491

2492
    """
2493
    valid_nodes = [node for node in self.cfg.GetOnlineNodeList()]
2494
    node_data = self.rpc.call_os_diagnose(valid_nodes)
2495
    pol = self._DiagnoseByOS(node_data)
2496
    output = []
2497
    calc_valid = self._FIELDS_NEEDVALID.intersection(self.op.output_fields)
2498
    calc_variants = "variants" in self.op.output_fields
2499

    
2500
    for os_name, os_data in pol.items():
2501
      row = []
2502
      if calc_valid:
2503
        valid = True
2504
        variants = None
2505
        for osl in os_data.values():
2506
          valid = valid and osl and osl[0][1]
2507
          if not valid:
2508
            variants = None
2509
            break
2510
          if calc_variants:
2511
            node_variants = osl[0][3]
2512
            if variants is None:
2513
              variants = node_variants
2514
            else:
2515
              variants = [v for v in variants if v in node_variants]
2516

    
2517
      for field in self.op.output_fields:
2518
        if field == "name":
2519
          val = os_name
2520
        elif field == "valid":
2521
          val = valid
2522
        elif field == "node_status":
2523
          # this is just a copy of the dict
2524
          val = {}
2525
          for node_name, nos_list in os_data.items():
2526
            val[node_name] = nos_list
2527
        elif field == "variants":
2528
          val =  variants
2529
        else:
2530
          raise errors.ParameterError(field)
2531
        row.append(val)
2532
      output.append(row)
2533

    
2534
    return output
2535

    
2536

    
2537
class LURemoveNode(LogicalUnit):
2538
  """Logical unit for removing a node.
2539

2540
  """
2541
  HPATH = "node-remove"
2542
  HTYPE = constants.HTYPE_NODE
2543
  _OP_REQP = ["node_name"]
2544

    
2545
  def BuildHooksEnv(self):
2546
    """Build hooks env.
2547

2548
    This doesn't run on the target node in the pre phase as a failed
2549
    node would then be impossible to remove.
2550

2551
    """
2552
    env = {
2553
      "OP_TARGET": self.op.node_name,
2554
      "NODE_NAME": self.op.node_name,
2555
      }
2556
    all_nodes = self.cfg.GetNodeList()
2557
    try:
2558
      all_nodes.remove(self.op.node_name)
2559
    except ValueError:
2560
      logging.warning("Node %s which is about to be removed not found"
2561
                      " in the all nodes list", self.op.node_name)
2562
    return env, all_nodes, all_nodes
2563

    
2564
  def CheckPrereq(self):
2565
    """Check prerequisites.
2566

2567
    This checks:
2568
     - the node exists in the configuration
2569
     - it does not have primary or secondary instances
2570
     - it's not the master
2571

2572
    Any errors are signaled by raising errors.OpPrereqError.
2573

2574
    """
2575
    self.op.node_name = _ExpandNodeName(self.cfg, self.op.node_name)
2576
    node = self.cfg.GetNodeInfo(self.op.node_name)
2577
    assert node is not None
2578

    
2579
    instance_list = self.cfg.GetInstanceList()
2580

    
2581
    masternode = self.cfg.GetMasterNode()
2582
    if node.name == masternode:
2583
      raise errors.OpPrereqError("Node is the master node,"
2584
                                 " you need to failover first.",
2585
                                 errors.ECODE_INVAL)
2586

    
2587
    for instance_name in instance_list:
2588
      instance = self.cfg.GetInstanceInfo(instance_name)
2589
      if node.name in instance.all_nodes:
2590
        raise errors.OpPrereqError("Instance %s is still running on the node,"
2591
                                   " please remove first." % instance_name,
2592
                                   errors.ECODE_INVAL)
2593
    self.op.node_name = node.name
2594
    self.node = node
2595

    
2596
  def Exec(self, feedback_fn):
2597
    """Removes the node from the cluster.
2598

2599
    """
2600
    node = self.node
2601
    logging.info("Stopping the node daemon and removing configs from node %s",
2602
                 node.name)
2603

    
2604
    modify_ssh_setup = self.cfg.GetClusterInfo().modify_ssh_setup
2605

    
2606
    # Promote nodes to master candidate as needed
2607
    _AdjustCandidatePool(self, exceptions=[node.name])
2608
    self.context.RemoveNode(node.name)
2609

    
2610
    # Run post hooks on the node before it's removed
2611
    hm = self.proc.hmclass(self.rpc.call_hooks_runner, self)
2612
    try:
2613
      hm.RunPhase(constants.HOOKS_PHASE_POST, [node.name])
2614
    except:
2615
      # pylint: disable-msg=W0702
2616
      self.LogWarning("Errors occurred running hooks on %s" % node.name)
2617

    
2618
    result = self.rpc.call_node_leave_cluster(node.name, modify_ssh_setup)
2619
    msg = result.fail_msg
2620
    if msg:
2621
      self.LogWarning("Errors encountered on the remote node while leaving"
2622
                      " the cluster: %s", msg)
2623

    
2624

    
2625
class LUQueryNodes(NoHooksLU):
2626
  """Logical unit for querying nodes.
2627

2628
  """
2629
  # pylint: disable-msg=W0142
2630
  _OP_REQP = ["output_fields", "names", "use_locking"]
2631
  REQ_BGL = False
2632

    
2633
  _SIMPLE_FIELDS = ["name", "serial_no", "ctime", "mtime", "uuid",
2634
                    "master_candidate", "offline", "drained"]
2635

    
2636
  _FIELDS_DYNAMIC = utils.FieldSet(
2637
    "dtotal", "dfree",
2638
    "mtotal", "mnode", "mfree",
2639
    "bootid",
2640
    "ctotal", "cnodes", "csockets",
2641
    )
2642

    
2643
  _FIELDS_STATIC = utils.FieldSet(*[
2644
    "pinst_cnt", "sinst_cnt",
2645
    "pinst_list", "sinst_list",
2646
    "pip", "sip", "tags",
2647
    "master",
2648
    "role"] + _SIMPLE_FIELDS
2649
    )
2650

    
2651
  def ExpandNames(self):
2652
    _CheckOutputFields(static=self._FIELDS_STATIC,
2653
                       dynamic=self._FIELDS_DYNAMIC,
2654
                       selected=self.op.output_fields)
2655

    
2656
    self.needed_locks = {}
2657
    self.share_locks[locking.LEVEL_NODE] = 1
2658

    
2659
    if self.op.names:
2660
      self.wanted = _GetWantedNodes(self, self.op.names)
2661
    else:
2662
      self.wanted = locking.ALL_SET
2663

    
2664
    self.do_node_query = self._FIELDS_STATIC.NonMatching(self.op.output_fields)
2665
    self.do_locking = self.do_node_query and self.op.use_locking
2666
    if self.do_locking:
2667
      # if we don't request only static fields, we need to lock the nodes
2668
      self.needed_locks[locking.LEVEL_NODE] = self.wanted
2669

    
2670
  def CheckPrereq(self):
2671
    """Check prerequisites.
2672

2673
    """
2674
    # The validation of the node list is done in the _GetWantedNodes,
2675
    # if non empty, and if empty, there's no validation to do
2676
    pass
2677

    
2678
  def Exec(self, feedback_fn):
2679
    """Computes the list of nodes and their attributes.
2680

2681
    """
2682
    all_info = self.cfg.GetAllNodesInfo()
2683
    if self.do_locking:
2684
      nodenames = self.acquired_locks[locking.LEVEL_NODE]
2685
    elif self.wanted != locking.ALL_SET:
2686
      nodenames = self.wanted
2687
      missing = set(nodenames).difference(all_info.keys())
2688
      if missing:
2689
        raise errors.OpExecError(
2690
          "Some nodes were removed before retrieving their data: %s" % missing)
2691
    else:
2692
      nodenames = all_info.keys()
2693

    
2694
    nodenames = utils.NiceSort(nodenames)
2695
    nodelist = [all_info[name] for name in nodenames]
2696

    
2697
    # begin data gathering
2698

    
2699
    if self.do_node_query:
2700
      live_data = {}
2701
      node_data = self.rpc.call_node_info(nodenames, self.cfg.GetVGName(),
2702
                                          self.cfg.GetHypervisorType())
2703
      for name in nodenames:
2704
        nodeinfo = node_data[name]
2705
        if not nodeinfo.fail_msg and nodeinfo.payload:
2706
          nodeinfo = nodeinfo.payload
2707
          fn = utils.TryConvert
2708
          live_data[name] = {
2709
            "mtotal": fn(int, nodeinfo.get('memory_total', None)),
2710
            "mnode": fn(int, nodeinfo.get('memory_dom0', None)),
2711
            "mfree": fn(int, nodeinfo.get('memory_free', None)),
2712
            "dtotal": fn(int, nodeinfo.get('vg_size', None)),
2713
            "dfree": fn(int, nodeinfo.get('vg_free', None)),
2714
            "ctotal": fn(int, nodeinfo.get('cpu_total', None)),
2715
            "bootid": nodeinfo.get('bootid', None),
2716
            "cnodes": fn(int, nodeinfo.get('cpu_nodes', None)),
2717
            "csockets": fn(int, nodeinfo.get('cpu_sockets', None)),
2718
            }
2719
        else:
2720
          live_data[name] = {}
2721
    else:
2722
      live_data = dict.fromkeys(nodenames, {})
2723

    
2724
    node_to_primary = dict([(name, set()) for name in nodenames])
2725
    node_to_secondary = dict([(name, set()) for name in nodenames])
2726

    
2727
    inst_fields = frozenset(("pinst_cnt", "pinst_list",
2728
                             "sinst_cnt", "sinst_list"))
2729
    if inst_fields & frozenset(self.op.output_fields):
2730
      inst_data = self.cfg.GetAllInstancesInfo()
2731

    
2732
      for inst in inst_data.values():
2733
        if inst.primary_node in node_to_primary:
2734
          node_to_primary[inst.primary_node].add(inst.name)
2735
        for secnode in inst.secondary_nodes:
2736
          if secnode in node_to_secondary:
2737
            node_to_secondary[secnode].add(inst.name)
2738

    
2739
    master_node = self.cfg.GetMasterNode()
2740

    
2741
    # end data gathering
2742

    
2743
    output = []
2744
    for node in nodelist:
2745
      node_output = []
2746
      for field in self.op.output_fields:
2747
        if field in self._SIMPLE_FIELDS:
2748
          val = getattr(node, field)
2749
        elif field == "pinst_list":
2750
          val = list(node_to_primary[node.name])
2751
        elif field == "sinst_list":
2752
          val = list(node_to_secondary[node.name])
2753
        elif field == "pinst_cnt":
2754
          val = len(node_to_primary[node.name])
2755
        elif field == "sinst_cnt":
2756
          val = len(node_to_secondary[node.name])
2757
        elif field == "pip":
2758
          val = node.primary_ip
2759
        elif field == "sip":
2760
          val = node.secondary_ip
2761
        elif field == "tags":
2762
          val = list(node.GetTags())
2763
        elif field == "master":
2764
          val = node.name == master_node
2765
        elif self._FIELDS_DYNAMIC.Matches(field):
2766
          val = live_data[node.name].get(field, None)
2767
        elif field == "role":
2768
          if node.name == master_node:
2769
            val = "M"
2770
          elif node.master_candidate:
2771
            val = "C"
2772
          elif node.drained:
2773
            val = "D"
2774
          elif node.offline:
2775
            val = "O"
2776
          else:
2777
            val = "R"
2778
        else:
2779
          raise errors.ParameterError(field)
2780
        node_output.append(val)
2781
      output.append(node_output)
2782

    
2783
    return output
2784

    
2785

    
2786
class LUQueryNodeVolumes(NoHooksLU):
2787
  """Logical unit for getting volumes on node(s).
2788

2789
  """
2790
  _OP_REQP = ["nodes", "output_fields"]
2791
  REQ_BGL = False
2792
  _FIELDS_DYNAMIC = utils.FieldSet("phys", "vg", "name", "size", "instance")
2793
  _FIELDS_STATIC = utils.FieldSet("node")
2794

    
2795
  def ExpandNames(self):
2796
    _CheckOutputFields(static=self._FIELDS_STATIC,
2797
                       dynamic=self._FIELDS_DYNAMIC,
2798
                       selected=self.op.output_fields)
2799

    
2800
    self.needed_locks = {}
2801
    self.share_locks[locking.LEVEL_NODE] = 1
2802
    if not self.op.nodes:
2803
      self.needed_locks[locking.LEVEL_NODE] = locking.ALL_SET
2804
    else:
2805
      self.needed_locks[locking.LEVEL_NODE] = \
2806
        _GetWantedNodes(self, self.op.nodes)
2807

    
2808
  def CheckPrereq(self):
2809
    """Check prerequisites.
2810

2811
    This checks that the fields required are valid output fields.
2812

2813
    """
2814
    self.nodes = self.acquired_locks[locking.LEVEL_NODE]
2815

    
2816
  def Exec(self, feedback_fn):
2817
    """Computes the list of nodes and their attributes.
2818

2819
    """
2820
    nodenames = self.nodes
2821
    volumes = self.rpc.call_node_volumes(nodenames)
2822

    
2823
    ilist = [self.cfg.GetInstanceInfo(iname) for iname
2824
             in self.cfg.GetInstanceList()]
2825

    
2826
    lv_by_node = dict([(inst, inst.MapLVsByNode()) for inst in ilist])
2827

    
2828
    output = []
2829
    for node in nodenames:
2830
      nresult = volumes[node]
2831
      if nresult.offline:
2832
        continue
2833
      msg = nresult.fail_msg
2834
      if msg:
2835
        self.LogWarning("Can't compute volume data on node %s: %s", node, msg)
2836
        continue
2837

    
2838
      node_vols = nresult.payload[:]
2839
      node_vols.sort(key=lambda vol: vol['dev'])
2840

    
2841
      for vol in node_vols:
2842
        node_output = []
2843
        for field in self.op.output_fields:
2844
          if field == "node":
2845
            val = node
2846
          elif field == "phys":
2847
            val = vol['dev']
2848
          elif field == "vg":
2849
            val = vol['vg']
2850
          elif field == "name":
2851
            val = vol['name']
2852
          elif field == "size":
2853
            val = int(float(vol['size']))
2854
          elif field == "instance":
2855
            for inst in ilist:
2856
              if node not in lv_by_node[inst]:
2857
                continue
2858
              if vol['name'] in lv_by_node[inst][node]:
2859
                val = inst.name
2860
                break
2861
            else:
2862
              val = '-'
2863
          else:
2864
            raise errors.ParameterError(field)
2865
          node_output.append(str(val))
2866

    
2867
        output.append(node_output)
2868

    
2869
    return output
2870

    
2871

    
2872
class LUQueryNodeStorage(NoHooksLU):
2873
  """Logical unit for getting information on storage units on node(s).
2874

2875
  """
2876
  _OP_REQP = ["nodes", "storage_type", "output_fields"]
2877
  REQ_BGL = False
2878
  _FIELDS_STATIC = utils.FieldSet(constants.SF_NODE)
2879

    
2880
  def ExpandNames(self):
2881
    storage_type = self.op.storage_type
2882

    
2883
    if storage_type not in constants.VALID_STORAGE_TYPES:
2884
      raise errors.OpPrereqError("Unknown storage type: %s" % storage_type,
2885
                                 errors.ECODE_INVAL)
2886

    
2887
    _CheckOutputFields(static=self._FIELDS_STATIC,
2888
                       dynamic=utils.FieldSet(*constants.VALID_STORAGE_FIELDS),
2889
                       selected=self.op.output_fields)
2890

    
2891
    self.needed_locks = {}
2892
    self.share_locks[locking.LEVEL_NODE] = 1
2893

    
2894
    if self.op.nodes:
2895
      self.needed_locks[locking.LEVEL_NODE] = \
2896
        _GetWantedNodes(self, self.op.nodes)
2897
    else:
2898
      self.needed_locks[locking.LEVEL_NODE] = locking.ALL_SET
2899

    
2900
  def CheckPrereq(self):
2901
    """Check prerequisites.
2902

2903
    This checks that the fields required are valid output fields.
2904

2905
    """
2906
    self.op.name = getattr(self.op, "name", None)
2907

    
2908
    self.nodes = self.acquired_locks[locking.LEVEL_NODE]
2909

    
2910
  def Exec(self, feedback_fn):
2911
    """Computes the list of nodes and their attributes.
2912

2913
    """
2914
    # Always get name to sort by
2915
    if constants.SF_NAME in self.op.output_fields:
2916
      fields = self.op.output_fields[:]
2917
    else:
2918
      fields = [constants.SF_NAME] + self.op.output_fields
2919

    
2920
    # Never ask for node or type as it's only known to the LU
2921
    for extra in [constants.SF_NODE, constants.SF_TYPE]:
2922
      while extra in fields:
2923
        fields.remove(extra)
2924

    
2925
    field_idx = dict([(name, idx) for (idx, name) in enumerate(fields)])
2926
    name_idx = field_idx[constants.SF_NAME]
2927

    
2928
    st_args = _GetStorageTypeArgs(self.cfg, self.op.storage_type)
2929
    data = self.rpc.call_storage_list(self.nodes,
2930
                                      self.op.storage_type, st_args,
2931
                                      self.op.name, fields)
2932

    
2933
    result = []
2934

    
2935
    for node in utils.NiceSort(self.nodes):
2936
      nresult = data[node]
2937
      if nresult.offline:
2938
        continue
2939

    
2940
      msg = nresult.fail_msg
2941
      if msg:
2942
        self.LogWarning("Can't get storage data from node %s: %s", node, msg)
2943
        continue
2944

    
2945
      rows = dict([(row[name_idx], row) for row in nresult.payload])
2946

    
2947
      for name in utils.NiceSort(rows.keys()):
2948
        row = rows[name]
2949

    
2950
        out = []
2951

    
2952
        for field in self.op.output_fields:
2953
          if field == constants.SF_NODE:
2954
            val = node
2955
          elif field == constants.SF_TYPE:
2956
            val = self.op.storage_type
2957
          elif field in field_idx:
2958
            val = row[field_idx[field]]
2959
          else:
2960
            raise errors.ParameterError(field)
2961

    
2962
          out.append(val)
2963

    
2964
        result.append(out)
2965

    
2966
    return result
2967

    
2968

    
2969
class LUModifyNodeStorage(NoHooksLU):
2970
  """Logical unit for modifying a storage volume on a node.
2971

2972
  """
2973
  _OP_REQP = ["node_name", "storage_type", "name", "changes"]
2974
  REQ_BGL = False
2975

    
2976
  def CheckArguments(self):
2977
    self.opnode_name = _ExpandNodeName(self.cfg, self.op.node_name)
2978

    
2979
    storage_type = self.op.storage_type
2980
    if storage_type not in constants.VALID_STORAGE_TYPES:
2981
      raise errors.OpPrereqError("Unknown storage type: %s" % storage_type,
2982
                                 errors.ECODE_INVAL)
2983

    
2984
  def ExpandNames(self):
2985
    self.needed_locks = {
2986
      locking.LEVEL_NODE: self.op.node_name,
2987
      }
2988

    
2989
  def CheckPrereq(self):
2990
    """Check prerequisites.
2991

2992
    """
2993
    storage_type = self.op.storage_type
2994

    
2995
    try:
2996
      modifiable = constants.MODIFIABLE_STORAGE_FIELDS[storage_type]
2997
    except KeyError:
2998
      raise errors.OpPrereqError("Storage units of type '%s' can not be"
2999
                                 " modified" % storage_type,
3000
                                 errors.ECODE_INVAL)
3001

    
3002
    diff = set(self.op.changes.keys()) - modifiable
3003
    if diff:
3004
      raise errors.OpPrereqError("The following fields can not be modified for"
3005
                                 " storage units of type '%s': %r" %
3006
                                 (storage_type, list(diff)),
3007
                                 errors.ECODE_INVAL)
3008

    
3009
  def Exec(self, feedback_fn):
3010
    """Computes the list of nodes and their attributes.
3011

3012
    """
3013
    st_args = _GetStorageTypeArgs(self.cfg, self.op.storage_type)
3014
    result = self.rpc.call_storage_modify(self.op.node_name,
3015
                                          self.op.storage_type, st_args,
3016
                                          self.op.name, self.op.changes)
3017
    result.Raise("Failed to modify storage unit '%s' on %s" %
3018
                 (self.op.name, self.op.node_name))
3019

    
3020

    
3021
class LUAddNode(LogicalUnit):
3022
  """Logical unit for adding node to the cluster.
3023

3024
  """
3025
  HPATH = "node-add"
3026
  HTYPE = constants.HTYPE_NODE
3027
  _OP_REQP = ["node_name"]
3028

    
3029
  def CheckArguments(self):
3030
    # validate/normalize the node name
3031
    self.op.node_name = utils.HostInfo.NormalizeName(self.op.node_name)
3032

    
3033
  def BuildHooksEnv(self):
3034
    """Build hooks env.
3035

3036
    This will run on all nodes before, and on all nodes + the new node after.
3037

3038
    """
3039
    env = {
3040
      "OP_TARGET": self.op.node_name,
3041
      "NODE_NAME": self.op.node_name,
3042
      "NODE_PIP": self.op.primary_ip,
3043
      "NODE_SIP": self.op.secondary_ip,
3044
      }
3045
    nodes_0 = self.cfg.GetNodeList()
3046
    nodes_1 = nodes_0 + [self.op.node_name, ]
3047
    return env, nodes_0, nodes_1
3048

    
3049
  def CheckPrereq(self):
3050
    """Check prerequisites.
3051

3052
    This checks:
3053
     - the new node is not already in the config
3054
     - it is resolvable
3055
     - its parameters (single/dual homed) matches the cluster
3056

3057
    Any errors are signaled by raising errors.OpPrereqError.
3058

3059
    """
3060
    node_name = self.op.node_name
3061
    cfg = self.cfg
3062

    
3063
    dns_data = utils.GetHostInfo(node_name)
3064

    
3065
    node = dns_data.name
3066
    primary_ip = self.op.primary_ip = dns_data.ip
3067
    secondary_ip = getattr(self.op, "secondary_ip", None)
3068
    if secondary_ip is None:
3069
      secondary_ip = primary_ip
3070
    if not utils.IsValidIP(secondary_ip):
3071
      raise errors.OpPrereqError("Invalid secondary IP given",
3072
                                 errors.ECODE_INVAL)
3073
    self.op.secondary_ip = secondary_ip
3074

    
3075
    node_list = cfg.GetNodeList()
3076
    if not self.op.readd and node in node_list:
3077
      raise errors.OpPrereqError("Node %s is already in the configuration" %
3078
                                 node, errors.ECODE_EXISTS)
3079
    elif self.op.readd and node not in node_list:
3080
      raise errors.OpPrereqError("Node %s is not in the configuration" % node,
3081
                                 errors.ECODE_NOENT)
3082

    
3083
    for existing_node_name in node_list:
3084
      existing_node = cfg.GetNodeInfo(existing_node_name)
3085

    
3086
      if self.op.readd and node == existing_node_name:
3087
        if (existing_node.primary_ip != primary_ip or
3088
            existing_node.secondary_ip != secondary_ip):
3089
          raise errors.OpPrereqError("Readded node doesn't have the same IP"
3090
                                     " address configuration as before",
3091
                                     errors.ECODE_INVAL)
3092
        continue
3093

    
3094
      if (existing_node.primary_ip == primary_ip or
3095
          existing_node.secondary_ip == primary_ip or
3096
          existing_node.primary_ip == secondary_ip or
3097
          existing_node.secondary_ip == secondary_ip):
3098
        raise errors.OpPrereqError("New node ip address(es) conflict with"
3099
                                   " existing node %s" % existing_node.name,
3100
                                   errors.ECODE_NOTUNIQUE)
3101

    
3102
    # check that the type of the node (single versus dual homed) is the
3103
    # same as for the master
3104
    myself = cfg.GetNodeInfo(self.cfg.GetMasterNode())
3105
    master_singlehomed = myself.secondary_ip == myself.primary_ip
3106
    newbie_singlehomed = secondary_ip == primary_ip
3107
    if master_singlehomed != newbie_singlehomed:
3108
      if master_singlehomed:
3109
        raise errors.OpPrereqError("The master has no private ip but the"
3110
                                   " new node has one",
3111
                                   errors.ECODE_INVAL)
3112
      else:
3113
        raise errors.OpPrereqError("The master has a private ip but the"
3114
                                   " new node doesn't have one",
3115
                                   errors.ECODE_INVAL)
3116

    
3117
    # checks reachability
3118
    if not utils.TcpPing(primary_ip, constants.DEFAULT_NODED_PORT):
3119
      raise errors.OpPrereqError("Node not reachable by ping",
3120
                                 errors.ECODE_ENVIRON)
3121

    
3122
    if not newbie_singlehomed:
3123
      # check reachability from my secondary ip to newbie's secondary ip
3124
      if not utils.TcpPing(secondary_ip, constants.DEFAULT_NODED_PORT,
3125
                           source=myself.secondary_ip):
3126
        raise errors.OpPrereqError("Node secondary ip not reachable by TCP"
3127
                                   " based ping to noded port",
3128
                                   errors.ECODE_ENVIRON)
3129

    
3130
    if self.op.readd:
3131
      exceptions = [node]
3132
    else:
3133
      exceptions = []
3134

    
3135
    self.master_candidate = _DecideSelfPromotion(self, exceptions=exceptions)
3136

    
3137
    if self.op.readd:
3138
      self.new_node = self.cfg.GetNodeInfo(node)
3139
      assert self.new_node is not None, "Can't retrieve locked node %s" % node
3140
    else:
3141
      self.new_node = objects.Node(name=node,
3142
                                   primary_ip=primary_ip,
3143
                                   secondary_ip=secondary_ip,
3144
                                   master_candidate=self.master_candidate,
3145
                                   offline=False, drained=False)
3146

    
3147
  def Exec(self, feedback_fn):
3148
    """Adds the new node to the cluster.
3149

3150
    """
3151
    new_node = self.new_node
3152
    node = new_node.name
3153

    
3154
    # for re-adds, reset the offline/drained/master-candidate flags;
3155
    # we need to reset here, otherwise offline would prevent RPC calls
3156
    # later in the procedure; this also means that if the re-add
3157
    # fails, we are left with a non-offlined, broken node
3158
    if self.op.readd:
3159
      new_node.drained = new_node.offline = False # pylint: disable-msg=W0201
3160
      self.LogInfo("Readding a node, the offline/drained flags were reset")
3161
      # if we demote the node, we do cleanup later in the procedure
3162
      new_node.master_candidate = self.master_candidate
3163

    
3164
    # notify the user about any possible mc promotion
3165
    if new_node.master_candidate:
3166
      self.LogInfo("Node will be a master candidate")
3167

    
3168
    # check connectivity
3169
    result = self.rpc.call_version([node])[node]
3170
    result.Raise("Can't get version information from node %s" % node)
3171
    if constants.PROTOCOL_VERSION == result.payload:
3172
      logging.info("Communication to node %s fine, sw version %s match",
3173
                   node, result.payload)
3174
    else:
3175
      raise errors.OpExecError("Version mismatch master version %s,"
3176
                               " node version %s" %
3177
                               (constants.PROTOCOL_VERSION, result.payload))
3178

    
3179
    # setup ssh on node
3180
    if self.cfg.GetClusterInfo().modify_ssh_setup:
3181
      logging.info("Copy ssh key to node %s", node)
3182
      priv_key, pub_key, _ = ssh.GetUserFiles(constants.GANETI_RUNAS)
3183
      keyarray = []
3184
      keyfiles = [constants.SSH_HOST_DSA_PRIV, constants.SSH_HOST_DSA_PUB,
3185
                  constants.SSH_HOST_RSA_PRIV, constants.SSH_HOST_RSA_PUB,
3186
                  priv_key, pub_key]
3187

    
3188
      for i in keyfiles:
3189
        keyarray.append(utils.ReadFile(i))
3190

    
3191
      result = self.rpc.call_node_add(node, keyarray[0], keyarray[1],
3192
                                      keyarray[2], keyarray[3], keyarray[4],
3193
                                      keyarray[5])
3194
      result.Raise("Cannot transfer ssh keys to the new node")
3195

    
3196
    # Add node to our /etc/hosts, and add key to known_hosts
3197
    if self.cfg.GetClusterInfo().modify_etc_hosts:
3198
      utils.AddHostToEtcHosts(new_node.name)
3199

    
3200
    if new_node.secondary_ip != new_node.primary_ip:
3201
      result = self.rpc.call_node_has_ip_address(new_node.name,
3202
                                                 new_node.secondary_ip)
3203
      result.Raise("Failure checking secondary ip on node %s" % new_node.name,
3204
                   prereq=True, ecode=errors.ECODE_ENVIRON)
3205
      if not result.payload:
3206
        raise errors.OpExecError("Node claims it doesn't have the secondary ip"
3207
                                 " you gave (%s). Please fix and re-run this"
3208
                                 " command." % new_node.secondary_ip)
3209

    
3210
    node_verify_list = [self.cfg.GetMasterNode()]
3211
    node_verify_param = {
3212
      constants.NV_NODELIST: [node],
3213
      # TODO: do a node-net-test as well?
3214
    }
3215

    
3216
    result = self.rpc.call_node_verify(node_verify_list, node_verify_param,
3217
                                       self.cfg.GetClusterName())
3218
    for verifier in node_verify_list:
3219
      result[verifier].Raise("Cannot communicate with node %s" % verifier)
3220
      nl_payload = result[verifier].payload[constants.NV_NODELIST]
3221
      if nl_payload:
3222
        for failed in nl_payload:
3223
          feedback_fn("ssh/hostname verification failed"
3224
                      " (checking from %s): %s" %
3225
                      (verifier, nl_payload[failed]))
3226
        raise errors.OpExecError("ssh/hostname verification failed.")
3227

    
3228
    if self.op.readd:
3229
      _RedistributeAncillaryFiles(self)
3230
      self.context.ReaddNode(new_node)
3231
      # make sure we redistribute the config
3232
      self.cfg.Update(new_node, feedback_fn)
3233
      # and make sure the new node will not have old files around
3234
      if not new_node.master_candidate:
3235
        result = self.rpc.call_node_demote_from_mc(new_node.name)
3236
        msg = result.fail_msg
3237
        if msg:
3238
          self.LogWarning("Node failed to demote itself from master"
3239
                          " candidate status: %s" % msg)
3240
    else:
3241
      _RedistributeAncillaryFiles(self, additional_nodes=[node])
3242
      self.context.AddNode(new_node, self.proc.GetECId())
3243

    
3244

    
3245
class LUSetNodeParams(LogicalUnit):
3246
  """Modifies the parameters of a node.
3247

3248
  """
3249
  HPATH = "node-modify"
3250
  HTYPE = constants.HTYPE_NODE
3251
  _OP_REQP = ["node_name"]
3252
  REQ_BGL = False
3253

    
3254
  def CheckArguments(self):
3255
    self.op.node_name = _ExpandNodeName(self.cfg, self.op.node_name)
3256
    _CheckBooleanOpField(self.op, 'master_candidate')
3257
    _CheckBooleanOpField(self.op, 'offline')
3258
    _CheckBooleanOpField(self.op, 'drained')
3259
    _CheckBooleanOpField(self.op, 'auto_promote')
3260
    all_mods = [self.op.offline, self.op.master_candidate, self.op.drained]
3261
    if all_mods.count(None) == 3:
3262
      raise errors.OpPrereqError("Please pass at least one modification",
3263
                                 errors.ECODE_INVAL)
3264
    if all_mods.count(True) > 1:
3265
      raise errors.OpPrereqError("Can't set the node into more than one"
3266
                                 " state at the same time",
3267
                                 errors.ECODE_INVAL)
3268

    
3269
    # Boolean value that tells us whether we're offlining or draining the node
3270
    self.offline_or_drain = (self.op.offline == True or
3271
                             self.op.drained == True)
3272
    self.deoffline_or_drain = (self.op.offline == False or
3273
                               self.op.drained == False)
3274
    self.might_demote = (self.op.master_candidate == False or
3275
                         self.offline_or_drain)
3276

    
3277
    self.lock_all = self.op.auto_promote and self.might_demote
3278

    
3279

    
3280
  def ExpandNames(self):
3281
    if self.lock_all:
3282
      self.needed_locks = {locking.LEVEL_NODE: locking.ALL_SET}
3283
    else:
3284
      self.needed_locks = {locking.LEVEL_NODE: self.op.node_name}
3285

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

3289
    This runs on the master node.
3290

3291
    """
3292
    env = {
3293
      "OP_TARGET": self.op.node_name,
3294
      "MASTER_CANDIDATE": str(self.op.master_candidate),
3295
      "OFFLINE": str(self.op.offline),
3296
      "DRAINED": str(self.op.drained),
3297
      }
3298
    nl = [self.cfg.GetMasterNode(),
3299
          self.op.node_name]
3300
    return env, nl, nl
3301

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

3305
    This only checks the instance list against the existing names.
3306

3307
    """
3308
    node = self.node = self.cfg.GetNodeInfo(self.op.node_name)
3309

    
3310
    if (self.op.master_candidate is not None or
3311
        self.op.drained is not None or
3312
        self.op.offline is not None):
3313
      # we can't change the master's node flags
3314
      if self.op.node_name == self.cfg.GetMasterNode():
3315
        raise errors.OpPrereqError("The master role can be changed"
3316
                                   " only via masterfailover",
3317
                                   errors.ECODE_INVAL)
3318

    
3319

    
3320
    if node.master_candidate and self.might_demote and not self.lock_all:
3321
      assert not self.op.auto_promote, "auto-promote set but lock_all not"
3322
      # check if after removing the current node, we're missing master
3323
      # candidates
3324
      (mc_remaining, mc_should, _) = \
3325
          self.cfg.GetMasterCandidateStats(exceptions=[node.name])
3326
      if mc_remaining != mc_should:
3327
        raise errors.OpPrereqError("Not enough master candidates, please"
3328
                                   " pass auto_promote to allow promotion",
3329
                                   errors.ECODE_INVAL)
3330

    
3331
    if (self.op.master_candidate == True and
3332
        ((node.offline and not self.op.offline == False) or
3333
         (node.drained and not self.op.drained == False))):
3334
      raise errors.OpPrereqError("Node '%s' is offline or drained, can't set"
3335
                                 " to master_candidate" % node.name,
3336
                                 errors.ECODE_INVAL)
3337

    
3338
    # If we're being deofflined/drained, we'll MC ourself if needed
3339
    if (self.deoffline_or_drain and not self.offline_or_drain and not
3340
        self.op.master_candidate == True and not node.master_candidate):
3341
      self.op.master_candidate = _DecideSelfPromotion(self)
3342
      if self.op.master_candidate:
3343
        self.LogInfo("Autopromoting node to master candidate")
3344

    
3345
    return
3346

    
3347
  def Exec(self, feedback_fn):
3348
    """Modifies a node.
3349

3350
    """
3351
    node = self.node
3352

    
3353
    result = []
3354
    changed_mc = False
3355

    
3356
    if self.op.offline is not None:
3357
      node.offline = self.op.offline
3358
      result.append(("offline", str(self.op.offline)))
3359
      if self.op.offline == True:
3360
        if node.master_candidate:
3361
          node.master_candidate = False
3362
          changed_mc = True
3363
          result.append(("master_candidate", "auto-demotion due to offline"))
3364
        if node.drained:
3365
          node.drained = False
3366
          result.append(("drained", "clear drained status due to offline"))
3367

    
3368
    if self.op.master_candidate is not None:
3369
      node.master_candidate = self.op.master_candidate
3370
      changed_mc = True
3371
      result.append(("master_candidate", str(self.op.master_candidate)))
3372
      if self.op.master_candidate == False:
3373
        rrc = self.rpc.call_node_demote_from_mc(node.name)
3374
        msg = rrc.fail_msg
3375
        if msg:
3376
          self.LogWarning("Node failed to demote itself: %s" % msg)
3377

    
3378
    if self.op.drained is not None:
3379
      node.drained = self.op.drained
3380
      result.append(("drained", str(self.op.drained)))
3381
      if self.op.drained == True:
3382
        if node.master_candidate:
3383
          node.master_candidate = False
3384
          changed_mc = True
3385
          result.append(("master_candidate", "auto-demotion due to drain"))
3386
          rrc = self.rpc.call_node_demote_from_mc(node.name)
3387
          msg = rrc.fail_msg
3388
          if msg:
3389
            self.LogWarning("Node failed to demote itself: %s" % msg)
3390
        if node.offline:
3391
          node.offline = False
3392
          result.append(("offline", "clear offline status due to drain"))
3393

    
3394
    # we locked all nodes, we adjust the CP before updating this node
3395
    if self.lock_all:
3396
      _AdjustCandidatePool(self, [node.name])
3397

    
3398
    # this will trigger configuration file update, if needed
3399
    self.cfg.Update(node, feedback_fn)
3400

    
3401
    # this will trigger job queue propagation or cleanup
3402
    if changed_mc:
3403
      self.context.ReaddNode(node)
3404

    
3405
    return result
3406

    
3407

    
3408
class LUPowercycleNode(NoHooksLU):
3409
  """Powercycles a node.
3410

3411
  """
3412
  _OP_REQP = ["node_name", "force"]
3413
  REQ_BGL = False
3414

    
3415
  def CheckArguments(self):
3416
    self.op.node_name = _ExpandNodeName(self.cfg, self.op.node_name)
3417
    if self.op.node_name == self.cfg.GetMasterNode() and not self.op.force:
3418
      raise errors.OpPrereqError("The node is the master and the force"
3419
                                 " parameter was not set",
3420
                                 errors.ECODE_INVAL)
3421

    
3422
  def ExpandNames(self):
3423
    """Locking for PowercycleNode.
3424

3425
    This is a last-resort option and shouldn't block on other
3426
    jobs. Therefore, we grab no locks.
3427

3428
    """
3429
    self.needed_locks = {}
3430

    
3431
  def CheckPrereq(self):
3432
    """Check prerequisites.
3433

3434
    This LU has no prereqs.
3435

3436
    """
3437
    pass
3438

    
3439
  def Exec(self, feedback_fn):
3440
    """Reboots a node.
3441

3442
    """
3443
    result = self.rpc.call_node_powercycle(self.op.node_name,
3444
                                           self.cfg.GetHypervisorType())
3445
    result.Raise("Failed to schedule the reboot")
3446
    return result.payload
3447

    
3448

    
3449
class LUQueryClusterInfo(NoHooksLU):
3450
  """Query cluster configuration.
3451

3452
  """
3453
  _OP_REQP = []
3454
  REQ_BGL = False
3455

    
3456
  def ExpandNames(self):
3457
    self.needed_locks = {}
3458

    
3459
  def CheckPrereq(self):
3460
    """No prerequsites needed for this LU.
3461

3462
    """
3463
    pass
3464

    
3465
  def Exec(self, feedback_fn):
3466
    """Return cluster config.
3467

3468
    """
3469
    cluster = self.cfg.GetClusterInfo()
3470
    os_hvp = {}
3471

    
3472
    # Filter just for enabled hypervisors
3473
    for os_name, hv_dict in cluster.os_hvp.items():
3474
      os_hvp[os_name] = {}
3475
      for hv_name, hv_params in hv_dict.items():
3476
        if hv_name in cluster.enabled_hypervisors:
3477
          os_hvp[os_name][hv_name] = hv_params
3478

    
3479
    result = {
3480
      "software_version": constants.RELEASE_VERSION,
3481
      "protocol_version": constants.PROTOCOL_VERSION,
3482
      "config_version": constants.CONFIG_VERSION,
3483
      "os_api_version": max(constants.OS_API_VERSIONS),
3484
      "export_version": constants.EXPORT_VERSION,
3485
      "architecture": (platform.architecture()[0], platform.machine()),
3486
      "name": cluster.cluster_name,
3487
      "master": cluster.master_node,
3488
      "default_hypervisor": cluster.enabled_hypervisors[0],
3489
      "enabled_hypervisors": cluster.enabled_hypervisors,
3490
      "hvparams": dict([(hypervisor_name, cluster.hvparams[hypervisor_name])
3491
                        for hypervisor_name in cluster.enabled_hypervisors]),
3492
      "os_hvp": os_hvp,
3493
      "beparams": cluster.beparams,
3494
      "nicparams": cluster.nicparams,
3495
      "candidate_pool_size": cluster.candidate_pool_size,
3496
      "master_netdev": cluster.master_netdev,
3497
      "volume_group_name": cluster.volume_group_name,
3498
      "file_storage_dir": cluster.file_storage_dir,
3499
      "ctime": cluster.ctime,
3500
      "mtime": cluster.mtime,
3501
      "uuid": cluster.uuid,
3502
      "tags": list(cluster.GetTags()),
3503
      }
3504

    
3505
    return result
3506

    
3507

    
3508
class LUQueryConfigValues(NoHooksLU):
3509
  """Return configuration values.
3510

3511
  """
3512
  _OP_REQP = []
3513
  REQ_BGL = False
3514
  _FIELDS_DYNAMIC = utils.FieldSet()
3515
  _FIELDS_STATIC = utils.FieldSet("cluster_name", "master_node", "drain_flag",
3516
                                  "watcher_pause")
3517

    
3518
  def ExpandNames(self):
3519
    self.needed_locks = {}
3520

    
3521
    _CheckOutputFields(static=self._FIELDS_STATIC,
3522
                       dynamic=self._FIELDS_DYNAMIC,
3523
                       selected=self.op.output_fields)
3524

    
3525
  def CheckPrereq(self):
3526
    """No prerequisites.
3527

3528
    """
3529
    pass
3530

    
3531
  def Exec(self, feedback_fn):
3532
    """Dump a representation of the cluster config to the standard output.
3533

3534
    """
3535
    values = []
3536
    for field in self.op.output_fields:
3537
      if field == "cluster_name":
3538
        entry = self.cfg.GetClusterName()
3539
      elif field == "master_node":
3540
        entry = self.cfg.GetMasterNode()
3541
      elif field == "drain_flag":
3542
        entry = os.path.exists(constants.JOB_QUEUE_DRAIN_FILE)
3543
      elif field == "watcher_pause":
3544
        entry = utils.ReadWatcherPauseFile(constants.WATCHER_PAUSEFILE)
3545
      else:
3546
        raise errors.ParameterError(field)
3547
      values.append(entry)
3548
    return values
3549

    
3550

    
3551
class LUActivateInstanceDisks(NoHooksLU):
3552
  """Bring up an instance's disks.
3553

3554
  """
3555
  _OP_REQP = ["instance_name"]
3556
  REQ_BGL = False
3557

    
3558
  def ExpandNames(self):
3559
    self._ExpandAndLockInstance()
3560
    self.needed_locks[locking.LEVEL_NODE] = []
3561
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
3562

    
3563
  def DeclareLocks(self, level):
3564
    if level == locking.LEVEL_NODE:
3565
      self._LockInstancesNodes()
3566

    
3567
  def CheckPrereq(self):
3568
    """Check prerequisites.
3569

3570
    This checks that the instance is in the cluster.
3571

3572
    """
3573
    self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
3574
    assert self.instance is not None, \
3575
      "Cannot retrieve locked instance %s" % self.op.instance_name
3576
    _CheckNodeOnline(self, self.instance.primary_node)
3577
    if not hasattr(self.op, "ignore_size"):
3578
      self.op.ignore_size = False
3579

    
3580
  def Exec(self, feedback_fn):
3581
    """Activate the disks.
3582

3583
    """
3584
    disks_ok, disks_info = \
3585
              _AssembleInstanceDisks(self, self.instance,
3586
                                     ignore_size=self.op.ignore_size)
3587
    if not disks_ok:
3588
      raise errors.OpExecError("Cannot activate block devices")
3589

    
3590
    return disks_info
3591

    
3592

    
3593
def _AssembleInstanceDisks(lu, instance, ignore_secondaries=False,
3594
                           ignore_size=False):
3595
  """Prepare the block devices for an instance.
3596

3597
  This sets up the block devices on all nodes.
3598

3599
  @type lu: L{LogicalUnit}
3600
  @param lu: the logical unit on whose behalf we execute
3601
  @type instance: L{objects.Instance}
3602
  @param instance: the instance for whose disks we assemble
3603
  @type ignore_secondaries: boolean
3604
  @param ignore_secondaries: if true, errors on secondary nodes
3605
      won't result in an error return from the function
3606
  @type ignore_size: boolean
3607
  @param ignore_size: if true, the current known size of the disk
3608
      will not be used during the disk activation, useful for cases
3609
      when the size is wrong
3610
  @return: False if the operation failed, otherwise a list of
3611
      (host, instance_visible_name, node_visible_name)
3612
      with the mapping from node devices to instance devices
3613

3614
  """
3615
  device_info = []
3616
  disks_ok = True
3617
  iname = instance.name
3618
  # With the two passes mechanism we try to reduce the window of
3619
  # opportunity for the race condition of switching DRBD to primary
3620
  # before handshaking occured, but we do not eliminate it
3621

    
3622
  # The proper fix would be to wait (with some limits) until the
3623
  # connection has been made and drbd transitions from WFConnection
3624
  # into any other network-connected state (Connected, SyncTarget,
3625
  # SyncSource, etc.)
3626

    
3627
  # 1st pass, assemble on all nodes in secondary mode
3628
  for inst_disk in instance.disks:
3629
    for node, node_disk in inst_disk.ComputeNodeTree(instance.primary_node):
3630
      if ignore_size:
3631
        node_disk = node_disk.Copy()
3632
        node_disk.UnsetSize()
3633
      lu.cfg.SetDiskID(node_disk, node)
3634
      result = lu.rpc.call_blockdev_assemble(node, node_disk, iname, False)
3635
      msg = result.fail_msg
3636
      if msg:
3637
        lu.proc.LogWarning("Could not prepare block device %s on node %s"
3638
                           " (is_primary=False, pass=1): %s",
3639
                           inst_disk.iv_name, node, msg)
3640
        if not ignore_secondaries:
3641
          disks_ok = False
3642

    
3643
  # FIXME: race condition on drbd migration to primary
3644

    
3645
  # 2nd pass, do only the primary node
3646
  for inst_disk in instance.disks:
3647
    dev_path = None
3648

    
3649
    for node, node_disk in inst_disk.ComputeNodeTree(instance.primary_node):
3650
      if node != instance.primary_node:
3651
        continue
3652
      if ignore_size:
3653
        node_disk = node_disk.Copy()
3654
        node_disk.UnsetSize()
3655
      lu.cfg.SetDiskID(node_disk, node)
3656
      result = lu.rpc.call_blockdev_assemble(node, node_disk, iname, True)
3657
      msg = result.fail_msg
3658
      if msg:
3659
        lu.proc.LogWarning("Could not prepare block device %s on node %s"
3660
                           " (is_primary=True, pass=2): %s",
3661
                           inst_disk.iv_name, node, msg)
3662
        disks_ok = False
3663
      else:
3664
        dev_path = result.payload
3665

    
3666
    device_info.append((instance.primary_node, inst_disk.iv_name, dev_path))
3667

    
3668
  # leave the disks configured for the primary node
3669
  # this is a workaround that would be fixed better by
3670
  # improving the logical/physical id handling
3671
  for disk in instance.disks:
3672
    lu.cfg.SetDiskID(disk, instance.primary_node)
3673

    
3674
  return disks_ok, device_info
3675

    
3676

    
3677
def _StartInstanceDisks(lu, instance, force):
3678
  """Start the disks of an instance.
3679

3680
  """
3681
  disks_ok, _ = _AssembleInstanceDisks(lu, instance,
3682
                                           ignore_secondaries=force)
3683
  if not disks_ok:
3684
    _ShutdownInstanceDisks(lu, instance)
3685
    if force is not None and not force:
3686
      lu.proc.LogWarning("", hint="If the message above refers to a"
3687
                         " secondary node,"
3688
                         " you can retry the operation using '--force'.")
3689
    raise errors.OpExecError("Disk consistency error")
3690

    
3691

    
3692
class LUDeactivateInstanceDisks(NoHooksLU):
3693
  """Shutdown an instance's disks.
3694

3695
  """
3696
  _OP_REQP = ["instance_name"]
3697
  REQ_BGL = False
3698

    
3699
  def ExpandNames(self):
3700
    self._ExpandAndLockInstance()
3701
    self.needed_locks[locking.LEVEL_NODE] = []
3702
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
3703

    
3704
  def DeclareLocks(self, level):
3705
    if level == locking.LEVEL_NODE:
3706
      self._LockInstancesNodes()
3707

    
3708
  def CheckPrereq(self):
3709
    """Check prerequisites.
3710

3711
    This checks that the instance is in the cluster.
3712

3713
    """
3714
    self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
3715
    assert self.instance is not None, \
3716
      "Cannot retrieve locked instance %s" % self.op.instance_name
3717

    
3718
  def Exec(self, feedback_fn):
3719
    """Deactivate the disks
3720

3721
    """
3722
    instance = self.instance
3723
    _SafeShutdownInstanceDisks(self, instance)
3724

    
3725

    
3726
def _SafeShutdownInstanceDisks(lu, instance):
3727
  """Shutdown block devices of an instance.
3728

3729
  This function checks if an instance is running, before calling
3730
  _ShutdownInstanceDisks.
3731

3732
  """
3733
  pnode = instance.primary_node
3734
  ins_l = lu.rpc.call_instance_list([pnode], [instance.hypervisor])[pnode]
3735
  ins_l.Raise("Can't contact node %s" % pnode)
3736

    
3737
  if instance.name in ins_l.payload:
3738
    raise errors.OpExecError("Instance is running, can't shutdown"
3739
                             " block devices.")
3740

    
3741
  _ShutdownInstanceDisks(lu, instance)
3742

    
3743

    
3744
def _ShutdownInstanceDisks(lu, instance, ignore_primary=False):
3745
  """Shutdown block devices of an instance.
3746

3747
  This does the shutdown on all nodes of the instance.
3748

3749
  If the ignore_primary is false, errors on the primary node are
3750
  ignored.
3751

3752
  """
3753
  all_result = True
3754
  for disk in instance.disks:
3755
    for node, top_disk in disk.ComputeNodeTree(instance.primary_node):
3756
      lu.cfg.SetDiskID(top_disk, node)
3757
      result = lu.rpc.call_blockdev_shutdown(node, top_disk)
3758
      msg = result.fail_msg
3759
      if msg:
3760
        lu.LogWarning("Could not shutdown block device %s on node %s: %s",
3761
                      disk.iv_name, node, msg)
3762
        if not ignore_primary or node != instance.primary_node:
3763
          all_result = False
3764
  return all_result
3765

    
3766

    
3767
def _CheckNodeFreeMemory(lu, node, reason, requested, hypervisor_name):
3768
  """Checks if a node has enough free memory.
3769

3770
  This function check if a given node has the needed amount of free
3771
  memory. In case the node has less memory or we cannot get the
3772
  information from the node, this function raise an OpPrereqError
3773
  exception.
3774

3775
  @type lu: C{LogicalUnit}
3776
  @param lu: a logical unit from which we get configuration data
3777
  @type node: C{str}
3778
  @param node: the node to check
3779
  @type reason: C{str}
3780
  @param reason: string to use in the error message
3781
  @type requested: C{int}
3782
  @param requested: the amount of memory in MiB to check for
3783
  @type hypervisor_name: C{str}
3784
  @param hypervisor_name: the hypervisor to ask for memory stats
3785
  @raise errors.OpPrereqError: if the node doesn't have enough memory, or
3786
      we cannot check the node
3787

3788
  """
3789
  nodeinfo = lu.rpc.call_node_info([node], lu.cfg.GetVGName(), hypervisor_name)
3790
  nodeinfo[node].Raise("Can't get data from node %s" % node,
3791
                       prereq=True, ecode=errors.ECODE_ENVIRON)
3792
  free_mem = nodeinfo[node].payload.get('memory_free', None)
3793
  if not isinstance(free_mem, int):
3794
    raise errors.OpPrereqError("Can't compute free memory on node %s, result"
3795
                               " was '%s'" % (node, free_mem),
3796
                               errors.ECODE_ENVIRON)
3797
  if requested > free_mem:
3798
    raise errors.OpPrereqError("Not enough memory on node %s for %s:"
3799
                               " needed %s MiB, available %s MiB" %
3800
                               (node, reason, requested, free_mem),
3801
                               errors.ECODE_NORES)
3802

    
3803

    
3804
class LUStartupInstance(LogicalUnit):
3805
  """Starts an instance.
3806

3807
  """
3808
  HPATH = "instance-start"
3809
  HTYPE = constants.HTYPE_INSTANCE
3810
  _OP_REQP = ["instance_name", "force"]
3811
  REQ_BGL = False
3812

    
3813
  def ExpandNames(self):
3814
    self._ExpandAndLockInstance()
3815

    
3816
  def BuildHooksEnv(self):
3817
    """Build hooks env.
3818

3819
    This runs on master, primary and secondary nodes of the instance.
3820

3821
    """
3822
    env = {
3823
      "FORCE": self.op.force,
3824
      }
3825
    env.update(_BuildInstanceHookEnvByObject(self, self.instance))
3826
    nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
3827
    return env, nl, nl
3828

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

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

3834
    """
3835
    self.instance = instance = self.cfg.GetInstanceInfo(self.op.instance_name)
3836
    assert self.instance is not None, \
3837
      "Cannot retrieve locked instance %s" % self.op.instance_name
3838

    
3839
    # extra beparams
3840
    self.beparams = getattr(self.op, "beparams", {})
3841
    if self.beparams:
3842
      if not isinstance(self.beparams, dict):
3843
        raise errors.OpPrereqError("Invalid beparams passed: %s, expected"
3844
                                   " dict" % (type(self.beparams), ),
3845
                                   errors.ECODE_INVAL)
3846
      # fill the beparams dict
3847
      utils.ForceDictType(self.beparams, constants.BES_PARAMETER_TYPES)
3848
      self.op.beparams = self.beparams
3849

    
3850
    # extra hvparams
3851
    self.hvparams = getattr(self.op, "hvparams", {})
3852
    if self.hvparams:
3853
      if not isinstance(self.hvparams, dict):
3854
        raise errors.OpPrereqError("Invalid hvparams passed: %s, expected"
3855
                                   " dict" % (type(self.hvparams), ),
3856
                                   errors.ECODE_INVAL)
3857

    
3858
      # check hypervisor parameter syntax (locally)
3859
      cluster = self.cfg.GetClusterInfo()
3860
      utils.ForceDictType(self.hvparams, constants.HVS_PARAMETER_TYPES)
3861
      filled_hvp = objects.FillDict(cluster.hvparams[instance.hypervisor],
3862
                                    instance.hvparams)
3863
      filled_hvp.update(self.hvparams)
3864
      hv_type = hypervisor.GetHypervisor(instance.hypervisor)
3865
      hv_type.CheckParameterSyntax(filled_hvp)
3866
      _CheckHVParams(self, instance.all_nodes, instance.hypervisor, filled_hvp)
3867
      self.op.hvparams = self.hvparams
3868

    
3869
    _CheckNodeOnline(self, instance.primary_node)
3870

    
3871
    bep = self.cfg.GetClusterInfo().FillBE(instance)
3872
    # check bridges existence
3873
    _CheckInstanceBridgesExist(self, instance)
3874

    
3875
    remote_info = self.rpc.call_instance_info(instance.primary_node,
3876
                                              instance.name,
3877
                                              instance.hypervisor)
3878
    remote_info.Raise("Error checking node %s" % instance.primary_node,
3879
                      prereq=True, ecode=errors.ECODE_ENVIRON)
3880
    if not remote_info.payload: # not running already
3881
      _CheckNodeFreeMemory(self, instance.primary_node,
3882
                           "starting instance %s" % instance.name,
3883
                           bep[constants.BE_MEMORY], instance.hypervisor)
3884

    
3885
  def Exec(self, feedback_fn):
3886
    """Start the instance.
3887

3888
    """
3889
    instance = self.instance
3890
    force = self.op.force
3891

    
3892
    self.cfg.MarkInstanceUp(instance.name)
3893

    
3894
    node_current = instance.primary_node
3895

    
3896
    _StartInstanceDisks(self, instance, force)
3897

    
3898
    result = self.rpc.call_instance_start(node_current, instance,
3899
                                          self.hvparams, self.beparams)
3900
    msg = result.fail_msg
3901
    if msg:
3902
      _ShutdownInstanceDisks(self, instance)
3903
      raise errors.OpExecError("Could not start instance: %s" % msg)
3904

    
3905

    
3906
class LURebootInstance(LogicalUnit):
3907
  """Reboot an instance.
3908

3909
  """
3910
  HPATH = "instance-reboot"
3911
  HTYPE = constants.HTYPE_INSTANCE
3912
  _OP_REQP = ["instance_name", "ignore_secondaries", "reboot_type"]
3913
  REQ_BGL = False
3914

    
3915
  def CheckArguments(self):
3916
    """Check the arguments.
3917

3918
    """
3919
    self.shutdown_timeout = getattr(self.op, "shutdown_timeout",
3920
                                    constants.DEFAULT_SHUTDOWN_TIMEOUT)
3921

    
3922
  def ExpandNames(self):
3923
    if self.op.reboot_type not in [constants.INSTANCE_REBOOT_SOFT,
3924
                                   constants.INSTANCE_REBOOT_HARD,
3925
                                   constants.INSTANCE_REBOOT_FULL]:
3926
      raise errors.ParameterError("reboot type not in [%s, %s, %s]" %
3927
                                  (constants.INSTANCE_REBOOT_SOFT,
3928
                                   constants.INSTANCE_REBOOT_HARD,
3929
                                   constants.INSTANCE_REBOOT_FULL))
3930
    self._ExpandAndLockInstance()
3931

    
3932
  def BuildHooksEnv(self):
3933
    """Build hooks env.
3934

3935
    This runs on master, primary and secondary nodes of the instance.
3936

3937
    """
3938
    env = {
3939
      "IGNORE_SECONDARIES": self.op.ignore_secondaries,
3940
      "REBOOT_TYPE": self.op.reboot_type,
3941
      "SHUTDOWN_TIMEOUT": self.shutdown_timeout,
3942
      }
3943
    env.update(_BuildInstanceHookEnvByObject(self, self.instance))
3944
    nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
3945
    return env, nl, nl
3946

    
3947
  def CheckPrereq(self):
3948
    """Check prerequisites.
3949

3950
    This checks that the instance is in the cluster.
3951

3952
    """
3953
    self.instance = instance = self.cfg.GetInstanceInfo(self.op.instance_name)
3954
    assert self.instance is not None, \
3955
      "Cannot retrieve locked instance %s" % self.op.instance_name
3956

    
3957
    _CheckNodeOnline(self, instance.primary_node)
3958

    
3959
    # check bridges existence
3960
    _CheckInstanceBridgesExist(self, instance)
3961

    
3962
  def Exec(self, feedback_fn):
3963
    """Reboot the instance.
3964

3965
    """
3966
    instance = self.instance
3967
    ignore_secondaries = self.op.ignore_secondaries
3968
    reboot_type = self.op.reboot_type
3969

    
3970
    node_current = instance.primary_node
3971

    
3972
    if reboot_type in [constants.INSTANCE_REBOOT_SOFT,
3973
                       constants.INSTANCE_REBOOT_HARD]:
3974
      for disk in instance.disks:
3975
        self.cfg.SetDiskID(disk, node_current)
3976
      result = self.rpc.call_instance_reboot(node_current, instance,
3977
                                             reboot_type,
3978
                                             self.shutdown_timeout)
3979
      result.Raise("Could not reboot instance")
3980
    else:
3981
      result = self.rpc.call_instance_shutdown(node_current, instance,
3982
                                               self.shutdown_timeout)
3983
      result.Raise("Could not shutdown instance for full reboot")
3984
      _ShutdownInstanceDisks(self, instance)
3985
      _StartInstanceDisks(self, instance, ignore_secondaries)
3986
      result = self.rpc.call_instance_start(node_current, instance, None, None)
3987
      msg = result.fail_msg
3988
      if msg:
3989
        _ShutdownInstanceDisks(self, instance)
3990
        raise errors.OpExecError("Could not start instance for"
3991
                                 " full reboot: %s" % msg)
3992

    
3993
    self.cfg.MarkInstanceUp(instance.name)
3994

    
3995

    
3996
class LUShutdownInstance(LogicalUnit):
3997
  """Shutdown an instance.
3998

3999
  """
4000
  HPATH = "instance-stop"
4001
  HTYPE = constants.HTYPE_INSTANCE
4002
  _OP_REQP = ["instance_name"]
4003
  REQ_BGL = False
4004

    
4005
  def CheckArguments(self):
4006
    """Check the arguments.
4007

4008
    """
4009
    self.timeout = getattr(self.op, "timeout",
4010
                           constants.DEFAULT_SHUTDOWN_TIMEOUT)
4011

    
4012
  def ExpandNames(self):
4013
    self._ExpandAndLockInstance()
4014

    
4015
  def BuildHooksEnv(self):
4016
    """Build hooks env.
4017

4018
    This runs on master, primary and secondary nodes of the instance.
4019

4020
    """
4021
    env = _BuildInstanceHookEnvByObject(self, self.instance)
4022
    env["TIMEOUT"] = self.timeout
4023
    nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
4024
    return env, nl, nl
4025

    
4026
  def CheckPrereq(self):
4027
    """Check prerequisites.
4028

4029
    This checks that the instance is in the cluster.
4030

4031
    """
4032
    self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
4033
    assert self.instance is not None, \
4034
      "Cannot retrieve locked instance %s" %