Statistics
| Branch: | Tag: | Revision:

root / lib / cmdlib.py @ ee844e20

History | View | Annotate | Download (312 kB)

1
#
2
#
3

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

    
21

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

    
24
# pylint: disable-msg=W0201
25

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

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

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

    
47

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

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

60
  Note that all commands require root permissions.
61

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

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

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

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

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

    
103
    # Tasklets
104
    self.tasklets = None
105

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

    
112
    self.CheckArguments()
113

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

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

    
122
  ssh = property(fget=__GetSSH)
123

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

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

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

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

139
    """
140
    pass
141

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

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

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

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

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

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

167
    Examples::
168

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

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

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

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

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

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

206
    """
207

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

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

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

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

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

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

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

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

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

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

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

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

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

265
    """
266
    raise NotImplementedError
267

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

    
357
    del self.recalculate_locks[locking.LEVEL_NODE]
358

    
359

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

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

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

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

373
    This just raises an error.
374

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

    
378

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

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

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

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

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

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

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

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

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

411
    """
412
    raise NotImplementedError
413

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

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

421
    """
422
    raise NotImplementedError
423

    
424

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

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

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

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

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

    
453
  return utils.NiceSort(wanted)
454

    
455

    
456
def _GetWantedInstances(lu, instances):
457
  """Returns list of checked and expanded instance names.
458

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

468
  """
469
  if not isinstance(instances, list):
470
    raise errors.OpPrereqError("Invalid argument type 'instances'",
471
                               errors.ECODE_INVAL)
472

    
473
  if instances:
474
    wanted = []
475

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

    
483
  else:
484
    wanted = utils.NiceSort(lu.cfg.GetInstanceList())
485
  return wanted
486

    
487

    
488
def _CheckOutputFields(static, dynamic, selected):
489
  """Checks whether all selected fields are valid.
490

491
  @type static: L{utils.FieldSet}
492
  @param static: static fields set
493
  @type dynamic: L{utils.FieldSet}
494
  @param dynamic: dynamic fields set
495

496
  """
497
  f = utils.FieldSet()
498
  f.Extend(static)
499
  f.Extend(dynamic)
500

    
501
  delta = f.NonMatching(selected)
502
  if delta:
503
    raise errors.OpPrereqError("Unknown output fields selected: %s"
504
                               % ",".join(delta), errors.ECODE_INVAL)
505

    
506

    
507
def _CheckBooleanOpField(op, name):
508
  """Validates boolean opcode parameters.
509

510
  This will ensure that an opcode parameter is either a boolean value,
511
  or None (but that it always exists).
512

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

    
520

    
521
def _CheckGlobalHvParams(params):
522
  """Validates that given hypervisor params are not global ones.
523

524
  This will ensure that instances don't get customised versions of
525
  global params.
526

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

    
535

    
536
def _CheckNodeOnline(lu, node):
537
  """Ensure that a given node is online.
538

539
  @param lu: the LU on behalf of which we make the check
540
  @param node: the node to check
541
  @raise errors.OpPrereqError: if the node is offline
542

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

    
548

    
549
def _CheckNodeNotDrained(lu, node):
550
  """Ensure that a given node is not drained.
551

552
  @param lu: the LU on behalf of which we make the check
553
  @param node: the node to check
554
  @raise errors.OpPrereqError: if the node is drained
555

556
  """
557
  if lu.cfg.GetNodeInfo(node).drained:
558
    raise errors.OpPrereqError("Can't use drained node %s" % node,
559
                               errors.ECODE_INVAL)
560

    
561

    
562
def _BuildInstanceHookEnv(name, primary_node, secondary_nodes, os_type, status,
563
                          memory, vcpus, nics, disk_template, disks,
564
                          bep, hvp, hypervisor_name):
565
  """Builds instance related env variables for hooks
566

567
  This builds the hook environment from individual variables.
568

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

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

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

    
631
  env["INSTANCE_NIC_COUNT"] = nic_count
632

    
633
  if disks:
634
    disk_count = len(disks)
635
    for idx, (size, mode) in enumerate(disks):
636
      env["INSTANCE_DISK%d_SIZE" % idx] = size
637
      env["INSTANCE_DISK%d_MODE" % idx] = mode
638
  else:
639
    disk_count = 0
640

    
641
  env["INSTANCE_DISK_COUNT"] = disk_count
642

    
643
  for source, kind in [(bep, "BE"), (hvp, "HV")]:
644
    for key, value in source.items():
645
      env["INSTANCE_%s_%s" % (kind, key)] = value
646

    
647
  return env
648

    
649

    
650
def _NICListToTuple(lu, nics):
651
  """Build a list of nic information tuples.
652

653
  This list is suitable to be passed to _BuildInstanceHookEnv or as a return
654
  value in LUQueryInstanceData.
655

656
  @type lu:  L{LogicalUnit}
657
  @param lu: the logical unit on whose behalf we execute
658
  @type nics: list of L{objects.NIC}
659
  @param nics: list of nics to convert to hooks tuples
660

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

    
673

    
674
def _BuildInstanceHookEnvByObject(lu, instance, override=None):
675
  """Builds instance related env variables for hooks from an object.
676

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

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

    
711

    
712
def _AdjustCandidatePool(lu, exceptions):
713
  """Adjust the candidate pool after node operations.
714

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

    
727

    
728
def _DecideSelfPromotion(lu, exceptions=None):
729
  """Decide whether I should promote myself as a master candidate.
730

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

    
738

    
739
def _CheckNicsBridgesExist(lu, target_nics, target_node,
740
                               profile=constants.PP_DEFAULT):
741
  """Check that the brigdes needed by a list of nics exist.
742

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

    
754

    
755
def _CheckInstanceBridgesExist(lu, instance, node=None):
756
  """Check that the brigdes needed by an instance exist.
757

758
  """
759
  if node is None:
760
    node = instance.primary_node
761
  _CheckNicsBridgesExist(lu, instance.nics, node)
762

    
763

    
764
def _CheckOSVariant(os_obj, name):
765
  """Check whether an OS name conforms to the os variants specification.
766

767
  @type os_obj: L{objects.OS}
768
  @param os_obj: OS object to check
769
  @type name: string
770
  @param name: OS name passed by the user, to check for validity
771

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

    
781
  if variant not in os_obj.supported_variants:
782
    raise errors.OpPrereqError("Unsupported OS variant", errors.ECODE_INVAL)
783

    
784

    
785
def _GetNodeInstancesInner(cfg, fn):
786
  return [i for i in cfg.GetAllInstancesInfo().values() if fn(i)]
787

    
788

    
789
def _GetNodeInstances(cfg, node_name):
790
  """Returns a list of all primary and secondary instances on a node.
791

792
  """
793

    
794
  return _GetNodeInstancesInner(cfg, lambda inst: node_name in inst.all_nodes)
795

    
796

    
797
def _GetNodePrimaryInstances(cfg, node_name):
798
  """Returns primary instances on a node.
799

800
  """
801
  return _GetNodeInstancesInner(cfg,
802
                                lambda inst: node_name == inst.primary_node)
803

    
804

    
805
def _GetNodeSecondaryInstances(cfg, node_name):
806
  """Returns secondary instances on a node.
807

808
  """
809
  return _GetNodeInstancesInner(cfg,
810
                                lambda inst: node_name in inst.secondary_nodes)
811

    
812

    
813
def _GetStorageTypeArgs(cfg, storage_type):
814
  """Returns the arguments for a storage type.
815

816
  """
817
  # Special case for file storage
818
  if storage_type == constants.ST_FILE:
819
    # storage.FileStorage wants a list of storage directories
820
    return [[cfg.GetFileStorageDir()]]
821

    
822
  return []
823

    
824

    
825
def _FindFaultyInstanceDisks(cfg, rpc, instance, node_name, prereq):
826
  faulty = []
827

    
828
  for dev in instance.disks:
829
    cfg.SetDiskID(dev, node_name)
830

    
831
  result = rpc.call_blockdev_getmirrorstatus(node_name, instance.disks)
832
  result.Raise("Failed to get disk status from node %s" % node_name,
833
               prereq=prereq, ecode=errors.ECODE_ENVIRON)
834

    
835
  for idx, bdev_status in enumerate(result.payload):
836
    if bdev_status and bdev_status.ldisk_status == constants.LDS_FAULTY:
837
      faulty.append(idx)
838

    
839
  return faulty
840

    
841

    
842
class LUPostInitCluster(LogicalUnit):
843
  """Logical unit for running hooks after cluster initialization.
844

845
  """
846
  HPATH = "cluster-init"
847
  HTYPE = constants.HTYPE_CLUSTER
848
  _OP_REQP = []
849

    
850
  def BuildHooksEnv(self):
851
    """Build hooks env.
852

853
    """
854
    env = {"OP_TARGET": self.cfg.GetClusterName()}
855
    mn = self.cfg.GetMasterNode()
856
    return env, [], [mn]
857

    
858
  def CheckPrereq(self):
859
    """No prerequisites to check.
860

861
    """
862
    return True
863

    
864
  def Exec(self, feedback_fn):
865
    """Nothing to do.
866

867
    """
868
    return True
869

    
870

    
871
class LUDestroyCluster(LogicalUnit):
872
  """Logical unit for destroying the cluster.
873

874
  """
875
  HPATH = "cluster-destroy"
876
  HTYPE = constants.HTYPE_CLUSTER
877
  _OP_REQP = []
878

    
879
  def BuildHooksEnv(self):
880
    """Build hooks env.
881

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

    
886
  def CheckPrereq(self):
887
    """Check prerequisites.
888

889
    This checks whether the cluster is empty.
890

891
    Any errors are signaled by raising errors.OpPrereqError.
892

893
    """
894
    master = self.cfg.GetMasterNode()
895

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

    
907
  def Exec(self, feedback_fn):
908
    """Destroys the cluster.
909

910
    """
911
    master = self.cfg.GetMasterNode()
912
    modify_ssh_setup = self.cfg.GetClusterInfo().modify_ssh_setup
913

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

    
922
    result = self.rpc.call_node_stop_master(master, False)
923
    result.Raise("Could not disable the master role")
924

    
925
    if modify_ssh_setup:
926
      priv_key, pub_key, _ = ssh.GetUserFiles(constants.GANETI_RUNAS)
927
      utils.CreateBackup(priv_key)
928
      utils.CreateBackup(pub_key)
929

    
930
    return master
931

    
932

    
933
class LUVerifyCluster(LogicalUnit):
934
  """Verifies the cluster status.
935

936
  """
937
  HPATH = "cluster-verify"
938
  HTYPE = constants.HTYPE_CLUSTER
939
  _OP_REQP = ["skip_checks", "verbose", "error_codes", "debug_simulate_errors"]
940
  REQ_BGL = False
941

    
942
  TCLUSTER = "cluster"
943
  TNODE = "node"
944
  TINSTANCE = "instance"
945

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

    
968
  ETYPE_FIELD = "code"
969
  ETYPE_ERROR = "ERROR"
970
  ETYPE_WARNING = "WARNING"
971

    
972
  def ExpandNames(self):
973
    self.needed_locks = {
974
      locking.LEVEL_NODE: locking.ALL_SET,
975
      locking.LEVEL_INSTANCE: locking.ALL_SET,
976
    }
977
    self.share_locks = dict.fromkeys(locking.LEVELS, 1)
978

    
979
  def _Error(self, ecode, item, msg, *args, **kwargs):
980
    """Format an error message.
981

982
    Based on the opcode's error_codes parameter, either format a
983
    parseable error code, or a simpler error string.
984

985
    This must be called only from Exec and functions called from Exec.
986

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

    
1005
  def _ErrorIf(self, cond, *args, **kwargs):
1006
    """Log an error message if the passed condition is True.
1007

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

    
1016
  def _VerifyNode(self, nodeinfo, file_list, local_cksum,
1017
                  node_result, master_files, drbd_map, vg_name):
1018
    """Run multiple tests against a node.
1019

1020
    Test list:
1021

1022
      - compares ganeti version
1023
      - checks vg existence and size > 20G
1024
      - checks config file checksum
1025
      - checks ssh to other nodes
1026

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

1038
    """
1039
    node = nodeinfo.name
1040
    _ErrorIf = self._ErrorIf # pylint: disable-msg=C0103
1041

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

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

    
1060
    test = local_version != remote_version[0]
1061
    _ErrorIf(test, self.ENODEVERSION, node,
1062
             "incompatible protocol versions: master %s,"
1063
             " node %s", local_version, remote_version[0])
1064
    if test:
1065
      return
1066

    
1067
    # node seems compatible, we can actually try to look into its results
1068

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

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

    
1086
    # checks config file checksum
1087

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

    
1115
    # checks ssh to any
1116

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

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

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

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

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

    
1179
  def _VerifyInstance(self, instance, instanceconfig, node_vol_is,
1180
                      node_instance, n_offline):
1181
    """Verify an instance.
1182

1183
    This function checks to see if the required block devices are
1184
    available on the instance's node.
1185

1186
    """
1187
    _ErrorIf = self._ErrorIf # pylint: disable-msg=C0103
1188
    node_current = instanceconfig.primary_node
1189

    
1190
    node_vol_should = {}
1191
    instanceconfig.MapLVsByNode(node_vol_should)
1192

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

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

    
1210
    for node in node_instance:
1211
      if (not node == node_current):
1212
        test = instance in node_instance[node]
1213
        _ErrorIf(test, self.EINSTANCEWRONGNODE, instance,
1214
                 "instance should not run on node %s", node)
1215

    
1216
  def _VerifyOrphanVolumes(self, node_vol_should, node_vol_is):
1217
    """Verify if there are any unknown volumes in the cluster.
1218

1219
    The .os, .swap and backup volumes are ignored. All other volumes are
1220
    reported as unknown.
1221

1222
    """
1223
    for node in node_vol_is:
1224
      for volume in node_vol_is[node]:
1225
        test = (node not in node_vol_should or
1226
                volume not in node_vol_should[node])
1227
        self._ErrorIf(test, self.ENODEORPHANLV, node,
1228
                      "volume %s is unknown", volume)
1229

    
1230
  def _VerifyOrphanInstances(self, instancelist, node_instance):
1231
    """Verify the list of running instances.
1232

1233
    This checks what instances are running but unknown to the cluster.
1234

1235
    """
1236
    for node in node_instance:
1237
      for o_inst in node_instance[node]:
1238
        test = o_inst not in instancelist
1239
        self._ErrorIf(test, self.ENODEORPHANINSTANCE, node,
1240
                      "instance %s on node %s should not exist", o_inst, node)
1241

    
1242
  def _VerifyNPlusOneMemory(self, node_info, instance_cfg):
1243
    """Verify N+1 Memory Resilience.
1244

1245
    Check that if one single node dies we can still start all the instances it
1246
    was primary for.
1247

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

    
1269
  def CheckPrereq(self):
1270
    """Check prerequisites.
1271

1272
    Transform the list of checks we're going to skip into a set and check that
1273
    all its members are valid.
1274

1275
    """
1276
    self.skip_set = frozenset(self.op.skip_checks)
1277
    if not constants.VERIFY_OPTIONAL_CHECKS.issuperset(self.skip_set):
1278
      raise errors.OpPrereqError("Invalid checks to be skipped specified",
1279
                                 errors.ECODE_INVAL)
1280

    
1281
  def BuildHooksEnv(self):
1282
    """Build hooks env.
1283

1284
    Cluster-Verify hooks just ran in the post phase and their failure makes
1285
    the output be logged in the verify output and the verification to fail.
1286

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

    
1295
    return env, [], all_nodes
1296

    
1297
  def Exec(self, feedback_fn):
1298
    """Verify integrity of cluster, performing various test on nodes.
1299

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

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

    
1325
    # FIXME: verify OS list
1326
    # do local checksums
1327
    master_files = [constants.CLUSTER_CONF_FILE]
1328

    
1329
    file_names = ssconf.SimpleStore().GetFileList()
1330
    file_names.append(constants.SSL_CERT_FILE)
1331
    file_names.append(constants.RAPI_CERT_FILE)
1332
    file_names.extend(master_files)
1333

    
1334
    local_checksums = utils.FingerprintFiles(file_names)
1335

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

    
1352
    if vg_name is not None:
1353
      node_verify_param[constants.NV_VGLIST] = None
1354
      node_verify_param[constants.NV_LVLIST] = vg_name
1355
      node_verify_param[constants.NV_PVLIST] = [vg_name]
1356
      node_verify_param[constants.NV_DRBDLIST] = None
1357

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

    
1367
    cluster = self.cfg.GetClusterInfo()
1368
    master_node = self.cfg.GetMasterNode()
1369
    all_drbd_map = self.cfg.ComputeDRBDMap()
1370

    
1371
    feedback_fn("* Verifying node status")
1372
    for node_i in nodeinfo:
1373
      node = node_i.name
1374

    
1375
      if node_i.offline:
1376
        if verbose:
1377
          feedback_fn("* Skipping offline node %s" % (node,))
1378
        n_offline.append(node)
1379
        continue
1380

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

    
1393
      msg = all_nvinfo[node].fail_msg
1394
      _ErrorIf(msg, self.ENODERPC, node, "while contacting node: %s", msg)
1395
      if msg:
1396
        continue
1397

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

    
1413
      self._VerifyNode(node_i, file_names, local_checksums,
1414
                       nresult, master_files, node_drbd, vg_name)
1415

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

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

    
1437
      node_instance[node] = idata
1438

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

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

    
1453
      if ntime_merged < (nvinfo_starttime - constants.NODE_MAX_CLOCK_SKEW):
1454
        ntime_diff = abs(nvinfo_starttime - ntime_merged)
1455
      elif ntime_merged > (nvinfo_endtime + constants.NODE_MAX_CLOCK_SKEW):
1456
        ntime_diff = abs(ntime_merged - nvinfo_endtime)
1457
      else:
1458
        ntime_diff = None
1459

    
1460
      _ErrorIf(ntime_diff is not None, self.ENODETIME, node,
1461
               "Node time diverges by at least %0.1fs from master node time",
1462
               ntime_diff)
1463

    
1464
      if ntime_diff is not None:
1465
        continue
1466

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

    
1495
    node_vol_should = {}
1496

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

    
1506
      inst_config.MapLVsByNode(node_vol_should)
1507

    
1508
      instance_cfg[instance] = inst_config
1509

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

    
1517
      if pnode in n_offline:
1518
        inst_nodes_offline.append(pnode)
1519

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

    
1531
      if not cluster.FillBE(inst_config)[constants.BE_AUTO_BALANCE]:
1532
        i_non_a_balanced.append(instance)
1533

    
1534
      for snode in inst_config.secondary_nodes:
1535
        _ErrorIf(snode not in node_info and snode not in n_offline,
1536
                 self.ENODERPC, snode,
1537
                 "instance %s, connection to secondary node"
1538
                 "failed", instance)
1539

    
1540
        if snode in node_info:
1541
          node_info[snode]['sinst'].append(instance)
1542
          if pnode not in node_info[snode]['sinst-by-pnode']:
1543
            node_info[snode]['sinst-by-pnode'][pnode] = []
1544
          node_info[snode]['sinst-by-pnode'][pnode].append(instance)
1545

    
1546
        if snode in n_offline:
1547
          inst_nodes_offline.append(snode)
1548

    
1549
      # warn that the instance lives on offline nodes
1550
      _ErrorIf(inst_nodes_offline, self.EINSTANCEBADNODE, instance,
1551
               "instance lives on offline node(s) %s",
1552
               utils.CommaJoin(inst_nodes_offline))
1553

    
1554
    feedback_fn("* Verifying orphan volumes")
1555
    self._VerifyOrphanVolumes(node_vol_should, node_volume)
1556

    
1557
    feedback_fn("* Verifying remaining instances")
1558
    self._VerifyOrphanInstances(instancelist, node_instance)
1559

    
1560
    if constants.VERIFY_NPLUSONE_MEM not in self.skip_set:
1561
      feedback_fn("* Verifying N+1 Memory redundancy")
1562
      self._VerifyNPlusOneMemory(node_info, instance_cfg)
1563

    
1564
    feedback_fn("* Other Notes")
1565
    if i_non_redundant:
1566
      feedback_fn("  - NOTICE: %d non-redundant instance(s) found."
1567
                  % len(i_non_redundant))
1568

    
1569
    if i_non_a_balanced:
1570
      feedback_fn("  - NOTICE: %d non-auto-balanced instance(s) found."
1571
                  % len(i_non_a_balanced))
1572

    
1573
    if n_offline:
1574
      feedback_fn("  - NOTICE: %d offline node(s) found." % len(n_offline))
1575

    
1576
    if n_drained:
1577
      feedback_fn("  - NOTICE: %d drained node(s) found." % len(n_drained))
1578

    
1579
    return not self.bad
1580

    
1581
  def HooksCallBack(self, phase, hooks_results, feedback_fn, lu_result):
1582
    """Analyze the post-hooks' result
1583

1584
    This method analyses the hook result, handles it, and sends some
1585
    nicely-formatted feedback back to the user.
1586

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

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

    
1604
      for node_name in hooks_results:
1605
        res = hooks_results[node_name]
1606
        msg = res.fail_msg
1607
        test = msg and not res.offline
1608
        self._ErrorIf(test, self.ENODEHOOKS, node_name,
1609
                      "Communication failure in hooks execution: %s", msg)
1610
        if res.offline or msg:
1611
          # No need to investigate payload if node is offline or gave an error.
1612
          # override manually lu_result here as _ErrorIf only
1613
          # overrides self.bad
1614
          lu_result = 1
1615
          continue
1616
        for script, hkr, output in res.payload:
1617
          test = hkr == constants.HKR_FAIL
1618
          self._ErrorIf(test, self.ENODEHOOKS, node_name,
1619
                        "Script %s failed, output:", script)
1620
          if test:
1621
            output = indent_re.sub('      ', output)
1622
            feedback_fn("%s" % output)
1623
            lu_result = 1
1624

    
1625
      return lu_result
1626

    
1627

    
1628
class LUVerifyDisks(NoHooksLU):
1629
  """Verifies the cluster disks status.
1630

1631
  """
1632
  _OP_REQP = []
1633
  REQ_BGL = False
1634

    
1635
  def ExpandNames(self):
1636
    self.needed_locks = {
1637
      locking.LEVEL_NODE: locking.ALL_SET,
1638
      locking.LEVEL_INSTANCE: locking.ALL_SET,
1639
    }
1640
    self.share_locks = dict.fromkeys(locking.LEVELS, 1)
1641

    
1642
  def CheckPrereq(self):
1643
    """Check prerequisites.
1644

1645
    This has no prerequisites.
1646

1647
    """
1648
    pass
1649

    
1650
  def Exec(self, feedback_fn):
1651
    """Verify integrity of cluster disks.
1652

1653
    @rtype: tuple of three items
1654
    @return: a tuple of (dict of node-to-node_error, list of instances
1655
        which need activate-disks, dict of instance: (node, volume) for
1656
        missing volumes
1657

1658
    """
1659
    result = res_nodes, res_instances, res_missing = {}, [], {}
1660

    
1661
    vg_name = self.cfg.GetVGName()
1662
    nodes = utils.NiceSort(self.cfg.GetNodeList())
1663
    instances = [self.cfg.GetInstanceInfo(name)
1664
                 for name in self.cfg.GetInstanceList()]
1665

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

    
1678
    if not nv_dict:
1679
      return result
1680

    
1681
    node_lvs = self.rpc.call_lv_list(nodes, vg_name)
1682

    
1683
    for node in nodes:
1684
      # node_volume
1685
      node_res = node_lvs[node]
1686
      if node_res.offline:
1687
        continue
1688
      msg = node_res.fail_msg
1689
      if msg:
1690
        logging.warning("Error enumerating LVs on node %s: %s", node, msg)
1691
        res_nodes[node] = msg
1692
        continue
1693

    
1694
      lvs = node_res.payload
1695
      for lv_name, (_, _, lv_online) in lvs.items():
1696
        inst = nv_dict.pop((node, lv_name), None)
1697
        if (not lv_online and inst is not None
1698
            and inst.name not in res_instances):
1699
          res_instances.append(inst.name)
1700

    
1701
    # any leftover items in nv_dict are missing LVs, let's arrange the
1702
    # data better
1703
    for key, inst in nv_dict.iteritems():
1704
      if inst.name not in res_missing:
1705
        res_missing[inst.name] = []
1706
      res_missing[inst.name].append(key)
1707

    
1708
    return result
1709

    
1710

    
1711
class LURepairDiskSizes(NoHooksLU):
1712
  """Verifies the cluster disks sizes.
1713

1714
  """
1715
  _OP_REQP = ["instances"]
1716
  REQ_BGL = False
1717

    
1718
  def ExpandNames(self):
1719
    if not isinstance(self.op.instances, list):
1720
      raise errors.OpPrereqError("Invalid argument type 'instances'",
1721
                                 errors.ECODE_INVAL)
1722

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

    
1744
  def DeclareLocks(self, level):
1745
    if level == locking.LEVEL_NODE and self.wanted_names is not None:
1746
      self._LockInstancesNodes(primary_only=True)
1747

    
1748
  def CheckPrereq(self):
1749
    """Check prerequisites.
1750

1751
    This only checks the optional instance list against the existing names.
1752

1753
    """
1754
    if self.wanted_names is None:
1755
      self.wanted_names = self.acquired_locks[locking.LEVEL_INSTANCE]
1756

    
1757
    self.wanted_instances = [self.cfg.GetInstanceInfo(name) for name
1758
                             in self.wanted_names]
1759

    
1760
  def _EnsureChildSizes(self, disk):
1761
    """Ensure children of the disk have the needed disk size.
1762

1763
    This is valid mainly for DRBD8 and fixes an issue where the
1764
    children have smaller disk size.
1765

1766
    @param disk: an L{ganeti.objects.Disk} object
1767

1768
    """
1769
    if disk.dev_type == constants.LD_DRBD8:
1770
      assert disk.children, "Empty children for DRBD8?"
1771
      fchild = disk.children[0]
1772
      mismatch = fchild.size < disk.size
1773
      if mismatch:
1774
        self.LogInfo("Child disk has size %d, parent %d, fixing",
1775
                     fchild.size, disk.size)
1776
        fchild.size = disk.size
1777

    
1778
      # and we recurse on this child only, not on the metadev
1779
      return self._EnsureChildSizes(fchild) or mismatch
1780
    else:
1781
      return False
1782

    
1783
  def Exec(self, feedback_fn):
1784
    """Verify the size of cluster disks.
1785

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

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

    
1833

    
1834
class LURenameCluster(LogicalUnit):
1835
  """Rename the cluster.
1836

1837
  """
1838
  HPATH = "cluster-rename"
1839
  HTYPE = constants.HTYPE_CLUSTER
1840
  _OP_REQP = ["name"]
1841

    
1842
  def BuildHooksEnv(self):
1843
    """Build hooks env.
1844

1845
    """
1846
    env = {
1847
      "OP_TARGET": self.cfg.GetClusterName(),
1848
      "NEW_NAME": self.op.name,
1849
      }
1850
    mn = self.cfg.GetMasterNode()
1851
    all_nodes = self.cfg.GetNodeList()
1852
    return env, [mn], all_nodes
1853

    
1854
  def CheckPrereq(self):
1855
    """Verify that the passed name is a valid one.
1856

1857
    """
1858
    hostname = utils.GetHostInfo(self.op.name)
1859

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

    
1874
    self.op.name = new_name
1875

    
1876
  def Exec(self, feedback_fn):
1877
    """Rename the cluster.
1878

1879
    """
1880
    clustername = self.op.name
1881
    ip = self.ip
1882

    
1883
    # shutdown the master IP
1884
    master = self.cfg.GetMasterNode()
1885
    result = self.rpc.call_node_stop_master(master, False)
1886
    result.Raise("Could not disable the master role")
1887

    
1888
    try:
1889
      cluster = self.cfg.GetClusterInfo()
1890
      cluster.cluster_name = clustername
1891
      cluster.master_ip = ip
1892
      self.cfg.Update(cluster, feedback_fn)
1893

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

    
1910
    finally:
1911
      result = self.rpc.call_node_start_master(master, False, False)
1912
      msg = result.fail_msg
1913
      if msg:
1914
        self.LogWarning("Could not re-enable the master role on"
1915
                        " the master, please restart manually: %s", msg)
1916

    
1917

    
1918
def _RecursiveCheckIfLVMBased(disk):
1919
  """Check if the given disk or its children are lvm-based.
1920

1921
  @type disk: L{objects.Disk}
1922
  @param disk: the disk to check
1923
  @rtype: boolean
1924
  @return: boolean indicating whether a LD_LV dev_type was found or not
1925

1926
  """
1927
  if disk.children:
1928
    for chdisk in disk.children:
1929
      if _RecursiveCheckIfLVMBased(chdisk):
1930
        return True
1931
  return disk.dev_type == constants.LD_LV
1932

    
1933

    
1934
class LUSetClusterParams(LogicalUnit):
1935
  """Change the parameters of the cluster.
1936

1937
  """
1938
  HPATH = "cluster-modify"
1939
  HTYPE = constants.HTYPE_CLUSTER
1940
  _OP_REQP = []
1941
  REQ_BGL = False
1942

    
1943
  def CheckArguments(self):
1944
    """Check parameters
1945

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

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

    
1967
  def BuildHooksEnv(self):
1968
    """Build hooks env.
1969

1970
    """
1971
    env = {
1972
      "OP_TARGET": self.cfg.GetClusterName(),
1973
      "NEW_VG_NAME": self.op.vg_name,
1974
      }
1975
    mn = self.cfg.GetMasterNode()
1976
    return env, [mn], [mn]
1977

    
1978
  def CheckPrereq(self):
1979
    """Check prerequisites.
1980

1981
    This checks whether the given params don't conflict and
1982
    if the given volume group is valid.
1983

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

    
1994
    node_list = self.acquired_locks[locking.LEVEL_NODE]
1995

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

    
2013
    self.cluster = cluster = self.cfg.GetClusterInfo()
2014
    # validate params changes
2015
    if self.op.beparams:
2016
      utils.ForceDictType(self.op.beparams, constants.BES_PARAMETER_TYPES)
2017
      self.new_beparams = objects.FillDict(
2018
        cluster.beparams[constants.PP_DEFAULT], self.op.beparams)
2019

    
2020
    if self.op.nicparams:
2021
      utils.ForceDictType(self.op.nicparams, constants.NICS_PARAMETER_TYPES)
2022
      self.new_nicparams = objects.FillDict(
2023
        cluster.nicparams[constants.PP_DEFAULT], self.op.nicparams)
2024
      objects.NIC.CheckParameterSyntax(self.new_nicparams)
2025
      nic_errors = []
2026

    
2027
      # check all instances for consistency
2028
      for instance in self.cfg.GetAllInstancesInfo().values():
2029
        for nic_idx, nic in enumerate(instance.nics):
2030
          params_copy = copy.deepcopy(nic.nicparams)
2031
          params_filled = objects.FillDict(self.new_nicparams, params_copy)
2032

    
2033
          # check parameter syntax
2034
          try:
2035
            objects.NIC.CheckParameterSyntax(params_filled)
2036
          except errors.ConfigurationError, err:
2037
            nic_errors.append("Instance %s, nic/%d: %s" %
2038
                              (instance.name, nic_idx, err))
2039

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

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

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

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

    
2088
  def Exec(self, feedback_fn):
2089
    """Change the parameters of the cluster.
2090

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

    
2110
    if self.op.candidate_pool_size is not None:
2111
      self.cluster.candidate_pool_size = self.op.candidate_pool_size
2112
      # we need to update the pool size here, otherwise the save will fail
2113
      _AdjustCandidatePool(self, [])
2114

    
2115
    self.cfg.Update(self.cluster, feedback_fn)
2116

    
2117

    
2118
def _RedistributeAncillaryFiles(lu, additional_nodes=None):
2119
  """Distribute additional files which are part of the cluster configuration.
2120

2121
  ConfigWriter takes care of distributing the config and ssconf files, but
2122
  there are more files which should be distributed to all nodes. This function
2123
  makes sure those are copied.
2124

2125
  @param lu: calling logical unit
2126
  @param additional_nodes: list of nodes not in the config to distribute to
2127

2128
  """
2129
  # 1. Gather target nodes
2130
  myself = lu.cfg.GetNodeInfo(lu.cfg.GetMasterNode())
2131
  dist_nodes = lu.cfg.GetNodeList()
2132
  if additional_nodes is not None:
2133
    dist_nodes.extend(additional_nodes)
2134
  if myself.name in dist_nodes:
2135
    dist_nodes.remove(myself.name)
2136

    
2137
  # 2. Gather files to distribute
2138
  dist_files = set([constants.ETC_HOSTS,
2139
                    constants.SSH_KNOWN_HOSTS_FILE,
2140
                    constants.RAPI_CERT_FILE,
2141
                    constants.RAPI_USERS_FILE,
2142
                    constants.HMAC_CLUSTER_KEY,
2143
                   ])
2144

    
2145
  enabled_hypervisors = lu.cfg.GetClusterInfo().enabled_hypervisors
2146
  for hv_name in enabled_hypervisors:
2147
    hv_class = hypervisor.GetHypervisor(hv_name)
2148
    dist_files.update(hv_class.GetAncillaryFiles())
2149

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

    
2161

    
2162
class LURedistributeConfig(NoHooksLU):
2163
  """Force the redistribution of cluster configuration.
2164

2165
  This is a very simple LU.
2166

2167
  """
2168
  _OP_REQP = []
2169
  REQ_BGL = False
2170

    
2171
  def ExpandNames(self):
2172
    self.needed_locks = {
2173
      locking.LEVEL_NODE: locking.ALL_SET,
2174
    }
2175
    self.share_locks[locking.LEVEL_NODE] = 1
2176

    
2177
  def CheckPrereq(self):
2178
    """Check prerequisites.
2179

2180
    """
2181

    
2182
  def Exec(self, feedback_fn):
2183
    """Redistribute the configuration.
2184

2185
    """
2186
    self.cfg.Update(self.cfg.GetClusterInfo(), feedback_fn)
2187
    _RedistributeAncillaryFiles(self)
2188

    
2189

    
2190
def _WaitForSync(lu, instance, oneshot=False):
2191
  """Sleep and poll for an instance's disk to sync.
2192

2193
  """
2194
  if not instance.disks:
2195
    return True
2196

    
2197
  if not oneshot:
2198
    lu.proc.LogInfo("Waiting for instance %s to sync disks." % instance.name)
2199

    
2200
  node = instance.primary_node
2201

    
2202
  for dev in instance.disks:
2203
    lu.cfg.SetDiskID(dev, node)
2204

    
2205
  # TODO: Convert to utils.Retry
2206

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

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

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

    
2253
    if done or oneshot:
2254
      break
2255

    
2256
    time.sleep(min(60, max_time))
2257

    
2258
  if done:
2259
    lu.proc.LogInfo("Instance %s's disks are in sync." % instance.name)
2260
  return not cumul_degraded
2261

    
2262

    
2263
def _CheckDiskConsistency(lu, dev, node, on_primary, ldisk=False):
2264
  """Check that mirrors are not degraded.
2265

2266
  The ldisk parameter, if True, will change the test from the
2267
  is_degraded attribute (which represents overall non-ok status for
2268
  the device(s)) to the ldisk (representing the local storage status).
2269

2270
  """
2271
  lu.cfg.SetDiskID(dev, node)
2272

    
2273
  result = True
2274

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

    
2290
  if dev.children:
2291
    for child in dev.children:
2292
      result = result and _CheckDiskConsistency(lu, child, node, on_primary)
2293

    
2294
  return result
2295

    
2296

    
2297
class LUDiagnoseOS(NoHooksLU):
2298
  """Logical unit for OS diagnose/query.
2299

2300
  """
2301
  _OP_REQP = ["output_fields", "names"]
2302
  REQ_BGL = False
2303
  _FIELDS_STATIC = utils.FieldSet()
2304
  _FIELDS_DYNAMIC = utils.FieldSet("name", "valid", "node_status", "variants")
2305
  # Fields that need calculation of global os validity
2306
  _FIELDS_NEEDVALID = frozenset(["valid", "variants"])
2307

    
2308
  def ExpandNames(self):
2309
    if self.op.names:
2310
      raise errors.OpPrereqError("Selective OS query not supported",
2311
                                 errors.ECODE_INVAL)
2312

    
2313
    _CheckOutputFields(static=self._FIELDS_STATIC,
2314
                       dynamic=self._FIELDS_DYNAMIC,
2315
                       selected=self.op.output_fields)
2316

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

    
2324
  def CheckPrereq(self):
2325
    """Check prerequisites.
2326

2327
    """
2328

    
2329
  @staticmethod
2330
  def _DiagnoseByOS(rlist):
2331
    """Remaps a per-node return list into an a per-os per-node dictionary
2332

2333
    @param rlist: a map with node names as keys and OS objects as values
2334

2335
    @rtype: dict
2336
    @return: a dictionary with osnames as keys and as value another map, with
2337
        nodes as keys and tuples of (path, status, diagnose) as values, eg::
2338

2339
          {"debian-etch": {"node1": [(/usr/lib/..., True, ""),
2340
                                     (/srv/..., False, "invalid api")],
2341
                           "node2": [(/srv/..., True, "")]}
2342
          }
2343

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

    
2364
  def Exec(self, feedback_fn):
2365
    """Compute the list of OSes.
2366

2367
    """
2368
    valid_nodes = [node for node in self.cfg.GetOnlineNodeList()]
2369
    node_data = self.rpc.call_os_diagnose(valid_nodes)
2370
    pol = self._DiagnoseByOS(node_data)
2371
    output = []
2372
    calc_valid = self._FIELDS_NEEDVALID.intersection(self.op.output_fields)
2373
    calc_variants = "variants" in self.op.output_fields
2374

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

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

    
2409
    return output
2410

    
2411

    
2412
class LURemoveNode(LogicalUnit):
2413
  """Logical unit for removing a node.
2414

2415
  """
2416
  HPATH = "node-remove"
2417
  HTYPE = constants.HTYPE_NODE
2418
  _OP_REQP = ["node_name"]
2419

    
2420
  def BuildHooksEnv(self):
2421
    """Build hooks env.
2422

2423
    This doesn't run on the target node in the pre phase as a failed
2424
    node would then be impossible to remove.
2425

2426
    """
2427
    env = {
2428
      "OP_TARGET": self.op.node_name,
2429
      "NODE_NAME": self.op.node_name,
2430
      }
2431
    all_nodes = self.cfg.GetNodeList()
2432
    try:
2433
      all_nodes.remove(self.op.node_name)
2434
    except ValueError:
2435
      logging.warning("Node %s which is about to be removed not found"
2436
                      " in the all nodes list", self.op.node_name)
2437
    return env, all_nodes, all_nodes
2438

    
2439
  def CheckPrereq(self):
2440
    """Check prerequisites.
2441

2442
    This checks:
2443
     - the node exists in the configuration
2444
     - it does not have primary or secondary instances
2445
     - it's not the master
2446

2447
    Any errors are signaled by raising errors.OpPrereqError.
2448

2449
    """
2450
    node = self.cfg.GetNodeInfo(self.cfg.ExpandNodeName(self.op.node_name))
2451
    if node is None:
2452
      raise errors.OpPrereqError("Node '%s' is unknown." % self.op.node_name,
2453
                                 errors.ECODE_NOENT)
2454

    
2455
    instance_list = self.cfg.GetInstanceList()
2456

    
2457
    masternode = self.cfg.GetMasterNode()
2458
    if node.name == masternode:
2459
      raise errors.OpPrereqError("Node is the master node,"
2460
                                 " you need to failover first.",
2461
                                 errors.ECODE_INVAL)
2462

    
2463
    for instance_name in instance_list:
2464
      instance = self.cfg.GetInstanceInfo(instance_name)
2465
      if node.name in instance.all_nodes:
2466
        raise errors.OpPrereqError("Instance %s is still running on the node,"
2467
                                   " please remove first." % instance_name,
2468
                                   errors.ECODE_INVAL)
2469
    self.op.node_name = node.name
2470
    self.node = node
2471

    
2472
  def Exec(self, feedback_fn):
2473
    """Removes the node from the cluster.
2474

2475
    """
2476
    node = self.node
2477
    logging.info("Stopping the node daemon and removing configs from node %s",
2478
                 node.name)
2479

    
2480
    modify_ssh_setup = self.cfg.GetClusterInfo().modify_ssh_setup
2481

    
2482
    # Promote nodes to master candidate as needed
2483
    _AdjustCandidatePool(self, exceptions=[node.name])
2484
    self.context.RemoveNode(node.name)
2485

    
2486
    # Run post hooks on the node before it's removed
2487
    hm = self.proc.hmclass(self.rpc.call_hooks_runner, self)
2488
    try:
2489
      hm.RunPhase(constants.HOOKS_PHASE_POST, [node.name])
2490
    except:
2491
      # pylint: disable-msg=W0702
2492
      self.LogWarning("Errors occurred running hooks on %s" % node.name)
2493

    
2494
    result = self.rpc.call_node_leave_cluster(node.name, modify_ssh_setup)
2495
    msg = result.fail_msg
2496
    if msg:
2497
      self.LogWarning("Errors encountered on the remote node while leaving"
2498
                      " the cluster: %s", msg)
2499

    
2500

    
2501
class LUQueryNodes(NoHooksLU):
2502
  """Logical unit for querying nodes.
2503

2504
  """
2505
  # pylint: disable-msg=W0142
2506
  _OP_REQP = ["output_fields", "names", "use_locking"]
2507
  REQ_BGL = False
2508

    
2509
  _SIMPLE_FIELDS = ["name", "serial_no", "ctime", "mtime", "uuid",
2510
                    "master_candidate", "offline", "drained"]
2511

    
2512
  _FIELDS_DYNAMIC = utils.FieldSet(
2513
    "dtotal", "dfree",
2514
    "mtotal", "mnode", "mfree",
2515
    "bootid",
2516
    "ctotal", "cnodes", "csockets",
2517
    )
2518

    
2519
  _FIELDS_STATIC = utils.FieldSet(*[
2520
    "pinst_cnt", "sinst_cnt",
2521
    "pinst_list", "sinst_list",
2522
    "pip", "sip", "tags",
2523
    "master",
2524
    "role"] + _SIMPLE_FIELDS
2525
    )
2526

    
2527
  def ExpandNames(self):
2528
    _CheckOutputFields(static=self._FIELDS_STATIC,
2529
                       dynamic=self._FIELDS_DYNAMIC,
2530
                       selected=self.op.output_fields)
2531

    
2532
    self.needed_locks = {}
2533
    self.share_locks[locking.LEVEL_NODE] = 1
2534

    
2535
    if self.op.names:
2536
      self.wanted = _GetWantedNodes(self, self.op.names)
2537
    else:
2538
      self.wanted = locking.ALL_SET
2539

    
2540
    self.do_node_query = self._FIELDS_STATIC.NonMatching(self.op.output_fields)
2541
    self.do_locking = self.do_node_query and self.op.use_locking
2542
    if self.do_locking:
2543
      # if we don't request only static fields, we need to lock the nodes
2544
      self.needed_locks[locking.LEVEL_NODE] = self.wanted
2545

    
2546
  def CheckPrereq(self):
2547
    """Check prerequisites.
2548

2549
    """
2550
    # The validation of the node list is done in the _GetWantedNodes,
2551
    # if non empty, and if empty, there's no validation to do
2552
    pass
2553

    
2554
  def Exec(self, feedback_fn):
2555
    """Computes the list of nodes and their attributes.
2556

2557
    """
2558
    all_info = self.cfg.GetAllNodesInfo()
2559
    if self.do_locking:
2560
      nodenames = self.acquired_locks[locking.LEVEL_NODE]
2561
    elif self.wanted != locking.ALL_SET:
2562
      nodenames = self.wanted
2563
      missing = set(nodenames).difference(all_info.keys())
2564
      if missing:
2565
        raise errors.OpExecError(
2566
          "Some nodes were removed before retrieving their data: %s" % missing)
2567
    else:
2568
      nodenames = all_info.keys()
2569

    
2570
    nodenames = utils.NiceSort(nodenames)
2571
    nodelist = [all_info[name] for name in nodenames]
2572

    
2573
    # begin data gathering
2574

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

    
2600
    node_to_primary = dict([(name, set()) for name in nodenames])
2601
    node_to_secondary = dict([(name, set()) for name in nodenames])
2602

    
2603
    inst_fields = frozenset(("pinst_cnt", "pinst_list",
2604
                             "sinst_cnt", "sinst_list"))
2605
    if inst_fields & frozenset(self.op.output_fields):
2606
      inst_data = self.cfg.GetAllInstancesInfo()
2607

    
2608
      for inst in inst_data.values():
2609
        if inst.primary_node in node_to_primary:
2610
          node_to_primary[inst.primary_node].add(inst.name)
2611
        for secnode in inst.secondary_nodes:
2612
          if secnode in node_to_secondary:
2613
            node_to_secondary[secnode].add(inst.name)
2614

    
2615
    master_node = self.cfg.GetMasterNode()
2616

    
2617
    # end data gathering
2618

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

    
2659
    return output
2660

    
2661

    
2662
class LUQueryNodeVolumes(NoHooksLU):
2663
  """Logical unit for getting volumes on node(s).
2664

2665
  """
2666
  _OP_REQP = ["nodes", "output_fields"]
2667
  REQ_BGL = False
2668
  _FIELDS_DYNAMIC = utils.FieldSet("phys", "vg", "name", "size", "instance")
2669
  _FIELDS_STATIC = utils.FieldSet("node")
2670

    
2671
  def ExpandNames(self):
2672
    _CheckOutputFields(static=self._FIELDS_STATIC,
2673
                       dynamic=self._FIELDS_DYNAMIC,
2674
                       selected=self.op.output_fields)
2675

    
2676
    self.needed_locks = {}
2677
    self.share_locks[locking.LEVEL_NODE] = 1
2678
    if not self.op.nodes:
2679
      self.needed_locks[locking.LEVEL_NODE] = locking.ALL_SET
2680
    else:
2681
      self.needed_locks[locking.LEVEL_NODE] = \
2682
        _GetWantedNodes(self, self.op.nodes)
2683

    
2684
  def CheckPrereq(self):
2685
    """Check prerequisites.
2686

2687
    This checks that the fields required are valid output fields.
2688

2689
    """
2690
    self.nodes = self.acquired_locks[locking.LEVEL_NODE]
2691

    
2692
  def Exec(self, feedback_fn):
2693
    """Computes the list of nodes and their attributes.
2694

2695
    """
2696
    nodenames = self.nodes
2697
    volumes = self.rpc.call_node_volumes(nodenames)
2698

    
2699
    ilist = [self.cfg.GetInstanceInfo(iname) for iname
2700
             in self.cfg.GetInstanceList()]
2701

    
2702
    lv_by_node = dict([(inst, inst.MapLVsByNode()) for inst in ilist])
2703

    
2704
    output = []
2705
    for node in nodenames:
2706
      nresult = volumes[node]
2707
      if nresult.offline:
2708
        continue
2709
      msg = nresult.fail_msg
2710
      if msg:
2711
        self.LogWarning("Can't compute volume data on node %s: %s", node, msg)
2712
        continue
2713

    
2714
      node_vols = nresult.payload[:]
2715
      node_vols.sort(key=lambda vol: vol['dev'])
2716

    
2717
      for vol in node_vols:
2718
        node_output = []
2719
        for field in self.op.output_fields:
2720
          if field == "node":
2721
            val = node
2722
          elif field == "phys":
2723
            val = vol['dev']
2724
          elif field == "vg":
2725
            val = vol['vg']
2726
          elif field == "name":
2727
            val = vol['name']
2728
          elif field == "size":
2729
            val = int(float(vol['size']))
2730
          elif field == "instance":
2731
            for inst in ilist:
2732
              if node not in lv_by_node[inst]:
2733
                continue
2734
              if vol['name'] in lv_by_node[inst][node]:
2735
                val = inst.name
2736
                break
2737
            else:
2738
              val = '-'
2739
          else:
2740
            raise errors.ParameterError(field)
2741
          node_output.append(str(val))
2742

    
2743
        output.append(node_output)
2744

    
2745
    return output
2746

    
2747

    
2748
class LUQueryNodeStorage(NoHooksLU):
2749
  """Logical unit for getting information on storage units on node(s).
2750

2751
  """
2752
  _OP_REQP = ["nodes", "storage_type", "output_fields"]
2753
  REQ_BGL = False
2754
  _FIELDS_STATIC = utils.FieldSet(constants.SF_NODE)
2755

    
2756
  def ExpandNames(self):
2757
    storage_type = self.op.storage_type
2758

    
2759
    if storage_type not in constants.VALID_STORAGE_TYPES:
2760
      raise errors.OpPrereqError("Unknown storage type: %s" % storage_type,
2761
                                 errors.ECODE_INVAL)
2762

    
2763
    _CheckOutputFields(static=self._FIELDS_STATIC,
2764
                       dynamic=utils.FieldSet(*constants.VALID_STORAGE_FIELDS),
2765
                       selected=self.op.output_fields)
2766

    
2767
    self.needed_locks = {}
2768
    self.share_locks[locking.LEVEL_NODE] = 1
2769

    
2770
    if self.op.nodes:
2771
      self.needed_locks[locking.LEVEL_NODE] = \
2772
        _GetWantedNodes(self, self.op.nodes)
2773
    else:
2774
      self.needed_locks[locking.LEVEL_NODE] = locking.ALL_SET
2775

    
2776
  def CheckPrereq(self):
2777
    """Check prerequisites.
2778

2779
    This checks that the fields required are valid output fields.
2780

2781
    """
2782
    self.op.name = getattr(self.op, "name", None)
2783

    
2784
    self.nodes = self.acquired_locks[locking.LEVEL_NODE]
2785

    
2786
  def Exec(self, feedback_fn):
2787
    """Computes the list of nodes and their attributes.
2788

2789
    """
2790
    # Always get name to sort by
2791
    if constants.SF_NAME in self.op.output_fields:
2792
      fields = self.op.output_fields[:]
2793
    else:
2794
      fields = [constants.SF_NAME] + self.op.output_fields
2795

    
2796
    # Never ask for node or type as it's only known to the LU
2797
    for extra in [constants.SF_NODE, constants.SF_TYPE]:
2798
      while extra in fields:
2799
        fields.remove(extra)
2800

    
2801
    field_idx = dict([(name, idx) for (idx, name) in enumerate(fields)])
2802
    name_idx = field_idx[constants.SF_NAME]
2803

    
2804
    st_args = _GetStorageTypeArgs(self.cfg, self.op.storage_type)
2805
    data = self.rpc.call_storage_list(self.nodes,
2806
                                      self.op.storage_type, st_args,
2807
                                      self.op.name, fields)
2808

    
2809
    result = []
2810

    
2811
    for node in utils.NiceSort(self.nodes):
2812
      nresult = data[node]
2813
      if nresult.offline:
2814
        continue
2815

    
2816
      msg = nresult.fail_msg
2817
      if msg:
2818
        self.LogWarning("Can't get storage data from node %s: %s", node, msg)
2819
        continue
2820

    
2821
      rows = dict([(row[name_idx], row) for row in nresult.payload])
2822

    
2823
      for name in utils.NiceSort(rows.keys()):
2824
        row = rows[name]
2825

    
2826
        out = []
2827

    
2828
        for field in self.op.output_fields:
2829
          if field == constants.SF_NODE:
2830
            val = node
2831
          elif field == constants.SF_TYPE:
2832
            val = self.op.storage_type
2833
          elif field in field_idx:
2834
            val = row[field_idx[field]]
2835
          else:
2836
            raise errors.ParameterError(field)
2837

    
2838
          out.append(val)
2839

    
2840
        result.append(out)
2841

    
2842
    return result
2843

    
2844

    
2845
class LUModifyNodeStorage(NoHooksLU):
2846
  """Logical unit for modifying a storage volume on a node.
2847

2848
  """
2849
  _OP_REQP = ["node_name", "storage_type", "name", "changes"]
2850
  REQ_BGL = False
2851

    
2852
  def CheckArguments(self):
2853
    node_name = self.cfg.ExpandNodeName(self.op.node_name)
2854
    if node_name is None:
2855
      raise errors.OpPrereqError("Invalid node name '%s'" % self.op.node_name,
2856
                                 errors.ECODE_NOENT)
2857

    
2858
    self.op.node_name = node_name
2859

    
2860
    storage_type = self.op.storage_type
2861
    if storage_type not in constants.VALID_STORAGE_TYPES:
2862
      raise errors.OpPrereqError("Unknown storage type: %s" % storage_type,
2863
                                 errors.ECODE_INVAL)
2864

    
2865
  def ExpandNames(self):
2866
    self.needed_locks = {
2867
      locking.LEVEL_NODE: self.op.node_name,
2868
      }
2869

    
2870
  def CheckPrereq(self):
2871
    """Check prerequisites.
2872

2873
    """
2874
    storage_type = self.op.storage_type
2875

    
2876
    try:
2877
      modifiable = constants.MODIFIABLE_STORAGE_FIELDS[storage_type]
2878
    except KeyError:
2879
      raise errors.OpPrereqError("Storage units of type '%s' can not be"
2880
                                 " modified" % storage_type,
2881
                                 errors.ECODE_INVAL)
2882

    
2883
    diff = set(self.op.changes.keys()) - modifiable
2884
    if diff:
2885
      raise errors.OpPrereqError("The following fields can not be modified for"
2886
                                 " storage units of type '%s': %r" %
2887
                                 (storage_type, list(diff)),
2888
                                 errors.ECODE_INVAL)
2889

    
2890
  def Exec(self, feedback_fn):
2891
    """Computes the list of nodes and their attributes.
2892

2893
    """
2894
    st_args = _GetStorageTypeArgs(self.cfg, self.op.storage_type)
2895
    result = self.rpc.call_storage_modify(self.op.node_name,
2896
                                          self.op.storage_type, st_args,
2897
                                          self.op.name, self.op.changes)
2898
    result.Raise("Failed to modify storage unit '%s' on %s" %
2899
                 (self.op.name, self.op.node_name))
2900

    
2901

    
2902
class LUAddNode(LogicalUnit):
2903
  """Logical unit for adding node to the cluster.
2904

2905
  """
2906
  HPATH = "node-add"
2907
  HTYPE = constants.HTYPE_NODE
2908
  _OP_REQP = ["node_name"]
2909

    
2910
  def BuildHooksEnv(self):
2911
    """Build hooks env.
2912

2913
    This will run on all nodes before, and on all nodes + the new node after.
2914

2915
    """
2916
    env = {
2917
      "OP_TARGET": self.op.node_name,
2918
      "NODE_NAME": self.op.node_name,
2919
      "NODE_PIP": self.op.primary_ip,
2920
      "NODE_SIP": self.op.secondary_ip,
2921
      }
2922
    nodes_0 = self.cfg.GetNodeList()
2923
    nodes_1 = nodes_0 + [self.op.node_name, ]
2924
    return env, nodes_0, nodes_1
2925

    
2926
  def CheckPrereq(self):
2927
    """Check prerequisites.
2928

2929
    This checks:
2930
     - the new node is not already in the config
2931
     - it is resolvable
2932
     - its parameters (single/dual homed) matches the cluster
2933

2934
    Any errors are signaled by raising errors.OpPrereqError.
2935

2936
    """
2937
    node_name = self.op.node_name
2938
    cfg = self.cfg
2939

    
2940
    dns_data = utils.GetHostInfo(node_name)
2941

    
2942
    node = dns_data.name
2943
    primary_ip = self.op.primary_ip = dns_data.ip
2944
    secondary_ip = getattr(self.op, "secondary_ip", None)
2945
    if secondary_ip is None:
2946
      secondary_ip = primary_ip
2947
    if not utils.IsValidIP(secondary_ip):
2948
      raise errors.OpPrereqError("Invalid secondary IP given",
2949
                                 errors.ECODE_INVAL)
2950
    self.op.secondary_ip = secondary_ip
2951

    
2952
    node_list = cfg.GetNodeList()
2953
    if not self.op.readd and node in node_list:
2954
      raise errors.OpPrereqError("Node %s is already in the configuration" %
2955
                                 node, errors.ECODE_EXISTS)
2956
    elif self.op.readd and node not in node_list:
2957
      raise errors.OpPrereqError("Node %s is not in the configuration" % node,
2958
                                 errors.ECODE_NOENT)
2959

    
2960
    for existing_node_name in node_list:
2961
      existing_node = cfg.GetNodeInfo(existing_node_name)
2962

    
2963
      if self.op.readd and node == existing_node_name:
2964
        if (existing_node.primary_ip != primary_ip or
2965
            existing_node.secondary_ip != secondary_ip):
2966
          raise errors.OpPrereqError("Readded node doesn't have the same IP"
2967
                                     " address configuration as before",
2968
                                     errors.ECODE_INVAL)
2969
        continue
2970

    
2971
      if (existing_node.primary_ip == primary_ip or
2972
          existing_node.secondary_ip == primary_ip or
2973
          existing_node.primary_ip == secondary_ip or
2974
          existing_node.secondary_ip == secondary_ip):
2975
        raise errors.OpPrereqError("New node ip address(es) conflict with"
2976
                                   " existing node %s" % existing_node.name,
2977
                                   errors.ECODE_NOTUNIQUE)
2978

    
2979
    # check that the type of the node (single versus dual homed) is the
2980
    # same as for the master
2981
    myself = cfg.GetNodeInfo(self.cfg.GetMasterNode())
2982
    master_singlehomed = myself.secondary_ip == myself.primary_ip
2983
    newbie_singlehomed = secondary_ip == primary_ip
2984
    if master_singlehomed != newbie_singlehomed:
2985
      if master_singlehomed:
2986
        raise errors.OpPrereqError("The master has no private ip but the"
2987
                                   " new node has one",
2988
                                   errors.ECODE_INVAL)
2989
      else:
2990
        raise errors.OpPrereqError("The master has a private ip but the"
2991
                                   " new node doesn't have one",
2992
                                   errors.ECODE_INVAL)
2993

    
2994
    # checks reachability
2995
    if not utils.TcpPing(primary_ip, constants.DEFAULT_NODED_PORT):
2996
      raise errors.OpPrereqError("Node not reachable by ping",
2997
                                 errors.ECODE_ENVIRON)
2998

    
2999
    if not newbie_singlehomed:
3000
      # check reachability from my secondary ip to newbie's secondary ip
3001
      if not utils.TcpPing(secondary_ip, constants.DEFAULT_NODED_PORT,
3002
                           source=myself.secondary_ip):
3003
        raise errors.OpPrereqError("Node secondary ip not reachable by TCP"
3004
                                   " based ping to noded port",
3005
                                   errors.ECODE_ENVIRON)
3006

    
3007
    if self.op.readd:
3008
      exceptions = [node]
3009
    else:
3010
      exceptions = []
3011

    
3012
    self.master_candidate = _DecideSelfPromotion(self, exceptions=exceptions)
3013

    
3014
    if self.op.readd:
3015
      self.new_node = self.cfg.GetNodeInfo(node)
3016
      assert self.new_node is not None, "Can't retrieve locked node %s" % node
3017
    else:
3018
      self.new_node = objects.Node(name=node,
3019
                                   primary_ip=primary_ip,
3020
                                   secondary_ip=secondary_ip,
3021
                                   master_candidate=self.master_candidate,
3022
                                   offline=False, drained=False)
3023

    
3024
  def Exec(self, feedback_fn):
3025
    """Adds the new node to the cluster.
3026

3027
    """
3028
    new_node = self.new_node
3029
    node = new_node.name
3030

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

    
3041
    # notify the user about any possible mc promotion
3042
    if new_node.master_candidate:
3043
      self.LogInfo("Node will be a master candidate")
3044

    
3045
    # check connectivity
3046
    result = self.rpc.call_version([node])[node]
3047
    result.Raise("Can't get version information from node %s" % node)
3048
    if constants.PROTOCOL_VERSION == result.payload:
3049
      logging.info("Communication to node %s fine, sw version %s match",
3050
                   node, result.payload)
3051
    else:
3052
      raise errors.OpExecError("Version mismatch master version %s,"
3053
                               " node version %s" %
3054
                               (constants.PROTOCOL_VERSION, result.payload))
3055

    
3056
    # setup ssh on node
3057
    if self.cfg.GetClusterInfo().modify_ssh_setup:
3058
      logging.info("Copy ssh key to node %s", node)
3059
      priv_key, pub_key, _ = ssh.GetUserFiles(constants.GANETI_RUNAS)
3060
      keyarray = []
3061
      keyfiles = [constants.SSH_HOST_DSA_PRIV, constants.SSH_HOST_DSA_PUB,
3062
                  constants.SSH_HOST_RSA_PRIV, constants.SSH_HOST_RSA_PUB,
3063
                  priv_key, pub_key]
3064

    
3065
      for i in keyfiles:
3066
        keyarray.append(utils.ReadFile(i))
3067

    
3068
      result = self.rpc.call_node_add(node, keyarray[0], keyarray[1],
3069
                                      keyarray[2], keyarray[3], keyarray[4],
3070
                                      keyarray[5])
3071
      result.Raise("Cannot transfer ssh keys to the new node")
3072

    
3073
    # Add node to our /etc/hosts, and add key to known_hosts
3074
    if self.cfg.GetClusterInfo().modify_etc_hosts:
3075
      utils.AddHostToEtcHosts(new_node.name)
3076

    
3077
    if new_node.secondary_ip != new_node.primary_ip:
3078
      result = self.rpc.call_node_has_ip_address(new_node.name,
3079
                                                 new_node.secondary_ip)
3080
      result.Raise("Failure checking secondary ip on node %s" % new_node.name,
3081
                   prereq=True, ecode=errors.ECODE_ENVIRON)
3082
      if not result.payload:
3083
        raise errors.OpExecError("Node claims it doesn't have the secondary ip"
3084
                                 " you gave (%s). Please fix and re-run this"
3085
                                 " command." % new_node.secondary_ip)
3086

    
3087
    node_verify_list = [self.cfg.GetMasterNode()]
3088
    node_verify_param = {
3089
      constants.NV_NODELIST: [node],
3090
      # TODO: do a node-net-test as well?
3091
    }
3092

    
3093
    result = self.rpc.call_node_verify(node_verify_list, node_verify_param,
3094
                                       self.cfg.GetClusterName())
3095
    for verifier in node_verify_list:
3096
      result[verifier].Raise("Cannot communicate with node %s" % verifier)
3097
      nl_payload = result[verifier].payload[constants.NV_NODELIST]
3098
      if nl_payload:
3099
        for failed in nl_payload:
3100
          feedback_fn("ssh/hostname verification failed"
3101
                      " (checking from %s): %s" %
3102
                      (verifier, nl_payload[failed]))
3103
        raise errors.OpExecError("ssh/hostname verification failed.")
3104

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

    
3121

    
3122
class LUSetNodeParams(LogicalUnit):
3123
  """Modifies the parameters of a node.
3124

3125
  """
3126
  HPATH = "node-modify"
3127
  HTYPE = constants.HTYPE_NODE
3128
  _OP_REQP = ["node_name"]
3129
  REQ_BGL = False
3130

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

    
3149
  def ExpandNames(self):
3150
    self.needed_locks = {locking.LEVEL_NODE: self.op.node_name}
3151

    
3152
  def BuildHooksEnv(self):
3153
    """Build hooks env.
3154

3155
    This runs on the master node.
3156

3157
    """
3158
    env = {
3159
      "OP_TARGET": self.op.node_name,
3160
      "MASTER_CANDIDATE": str(self.op.master_candidate),
3161
      "OFFLINE": str(self.op.offline),
3162
      "DRAINED": str(self.op.drained),
3163
      }
3164
    nl = [self.cfg.GetMasterNode(),
3165
          self.op.node_name]
3166
    return env, nl, nl
3167

    
3168
  def CheckPrereq(self):
3169
    """Check prerequisites.
3170

3171
    This only checks the instance list against the existing names.
3172

3173
    """
3174
    node = self.node = self.cfg.GetNodeInfo(self.op.node_name)
3175

    
3176
    if (self.op.master_candidate is not None or
3177
        self.op.drained is not None or
3178
        self.op.offline is not None):
3179
      # we can't change the master's node flags
3180
      if self.op.node_name == self.cfg.GetMasterNode():
3181
        raise errors.OpPrereqError("The master role can be changed"
3182
                                   " only via masterfailover",
3183
                                   errors.ECODE_INVAL)
3184

    
3185
    # Boolean value that tells us whether we're offlining or draining the node
3186
    offline_or_drain = self.op.offline == True or self.op.drained == True
3187
    deoffline_or_drain = self.op.offline == False or self.op.drained == False
3188

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

    
3205
    if (self.op.master_candidate == True and
3206
        ((node.offline and not self.op.offline == False) or
3207
         (node.drained and not self.op.drained == False))):
3208
      raise errors.OpPrereqError("Node '%s' is offline or drained, can't set"
3209
                                 " to master_candidate" % node.name,
3210
                                 errors.ECODE_INVAL)
3211

    
3212
    # If we're being deofflined/drained, we'll MC ourself if needed
3213
    if (deoffline_or_drain and not offline_or_drain and not
3214
        self.op.master_candidate == True and not node.master_candidate):
3215
      self.op.master_candidate = _DecideSelfPromotion(self)
3216
      if self.op.master_candidate:
3217
        self.LogInfo("Autopromoting node to master candidate")
3218

    
3219
    return
3220

    
3221
  def Exec(self, feedback_fn):
3222
    """Modifies a node.
3223

3224
    """
3225
    node = self.node
3226

    
3227
    result = []
3228
    changed_mc = False
3229

    
3230
    if self.op.offline is not None:
3231
      node.offline = self.op.offline
3232
      result.append(("offline", str(self.op.offline)))
3233
      if self.op.offline == True:
3234
        if node.master_candidate:
3235
          node.master_candidate = False
3236
          changed_mc = True
3237
          result.append(("master_candidate", "auto-demotion due to offline"))
3238
        if node.drained:
3239
          node.drained = False
3240
          result.append(("drained", "clear drained status due to offline"))
3241

    
3242
    if self.op.master_candidate is not None:
3243
      node.master_candidate = self.op.master_candidate
3244
      changed_mc = True
3245
      result.append(("master_candidate", str(self.op.master_candidate)))
3246
      if self.op.master_candidate == False:
3247
        rrc = self.rpc.call_node_demote_from_mc(node.name)
3248
        msg = rrc.fail_msg
3249
        if msg:
3250
          self.LogWarning("Node failed to demote itself: %s" % msg)
3251

    
3252
    if self.op.drained is not None:
3253
      node.drained = self.op.drained
3254
      result.append(("drained", str(self.op.drained)))
3255
      if self.op.drained == True:
3256
        if node.master_candidate:
3257
          node.master_candidate = False
3258
          changed_mc = True
3259
          result.append(("master_candidate", "auto-demotion due to drain"))
3260
          rrc = self.rpc.call_node_demote_from_mc(node.name)
3261
          msg = rrc.fail_msg
3262
          if msg:
3263
            self.LogWarning("Node failed to demote itself: %s" % msg)
3264
        if node.offline:
3265
          node.offline = False
3266
          result.append(("offline", "clear offline status due to drain"))
3267

    
3268
    # this will trigger configuration file update, if needed
3269
    self.cfg.Update(node, feedback_fn)
3270
    # this will trigger job queue propagation or cleanup
3271
    if changed_mc:
3272
      self.context.ReaddNode(node)
3273

    
3274
    return result
3275

    
3276

    
3277
class LUPowercycleNode(NoHooksLU):
3278
  """Powercycles a node.
3279

3280
  """
3281
  _OP_REQP = ["node_name", "force"]
3282
  REQ_BGL = False
3283

    
3284
  def CheckArguments(self):
3285
    node_name = self.cfg.ExpandNodeName(self.op.node_name)
3286
    if node_name is None:
3287
      raise errors.OpPrereqError("Invalid node name '%s'" % self.op.node_name,
3288
                                 errors.ECODE_NOENT)
3289
    self.op.node_name = node_name
3290
    if node_name == self.cfg.GetMasterNode() and not self.op.force:
3291
      raise errors.OpPrereqError("The node is the master and the force"
3292
                                 " parameter was not set",
3293
                                 errors.ECODE_INVAL)
3294

    
3295
  def ExpandNames(self):
3296
    """Locking for PowercycleNode.
3297

3298
    This is a last-resort option and shouldn't block on other
3299
    jobs. Therefore, we grab no locks.
3300

3301
    """
3302
    self.needed_locks = {}
3303

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

3307
    This LU has no prereqs.
3308

3309
    """
3310
    pass
3311

    
3312
  def Exec(self, feedback_fn):
3313
    """Reboots a node.
3314

3315
    """
3316
    result = self.rpc.call_node_powercycle(self.op.node_name,
3317
                                           self.cfg.GetHypervisorType())
3318
    result.Raise("Failed to schedule the reboot")
3319
    return result.payload
3320

    
3321

    
3322
class LUQueryClusterInfo(NoHooksLU):
3323
  """Query cluster configuration.
3324

3325
  """
3326
  _OP_REQP = []
3327
  REQ_BGL = False
3328

    
3329
  def ExpandNames(self):
3330
    self.needed_locks = {}
3331

    
3332
  def CheckPrereq(self):
3333
    """No prerequsites needed for this LU.
3334

3335
    """
3336
    pass
3337

    
3338
  def Exec(self, feedback_fn):
3339
    """Return cluster config.
3340

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

    
3368
    return result
3369

    
3370

    
3371
class LUQueryConfigValues(NoHooksLU):
3372
  """Return configuration values.
3373

3374
  """
3375
  _OP_REQP = []
3376
  REQ_BGL = False
3377
  _FIELDS_DYNAMIC = utils.FieldSet()
3378
  _FIELDS_STATIC = utils.FieldSet("cluster_name", "master_node", "drain_flag",
3379
                                  "watcher_pause")
3380

    
3381
  def ExpandNames(self):
3382
    self.needed_locks = {}
3383

    
3384
    _CheckOutputFields(static=self._FIELDS_STATIC,
3385
                       dynamic=self._FIELDS_DYNAMIC,
3386
                       selected=self.op.output_fields)
3387

    
3388
  def CheckPrereq(self):
3389
    """No prerequisites.
3390

3391
    """
3392
    pass
3393

    
3394
  def Exec(self, feedback_fn):
3395
    """Dump a representation of the cluster config to the standard output.
3396

3397
    """
3398
    values = []
3399
    for field in self.op.output_fields:
3400
      if field == "cluster_name":
3401
        entry = self.cfg.GetClusterName()
3402
      elif field == "master_node":
3403
        entry = self.cfg.GetMasterNode()
3404
      elif field == "drain_flag":
3405
        entry = os.path.exists(constants.JOB_QUEUE_DRAIN_FILE)
3406
      elif field == "watcher_pause":
3407
        return utils.ReadWatcherPauseFile(constants.WATCHER_PAUSEFILE)
3408
      else:
3409
        raise errors.ParameterError(field)
3410
      values.append(entry)
3411
    return values
3412

    
3413

    
3414
class LUActivateInstanceDisks(NoHooksLU):
3415
  """Bring up an instance's disks.
3416

3417
  """
3418
  _OP_REQP = ["instance_name"]
3419
  REQ_BGL = False
3420

    
3421
  def ExpandNames(self):
3422
    self._ExpandAndLockInstance()
3423
    self.needed_locks[locking.LEVEL_NODE] = []
3424
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
3425

    
3426
  def DeclareLocks(self, level):
3427
    if level == locking.LEVEL_NODE:
3428
      self._LockInstancesNodes()
3429

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

3433
    This checks that the instance is in the cluster.
3434

3435
    """
3436
    self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
3437
    assert self.instance is not None, \
3438
      "Cannot retrieve locked instance %s" % self.op.instance_name
3439
    _CheckNodeOnline(self, self.instance.primary_node)
3440
    if not hasattr(self.op, "ignore_size"):
3441
      self.op.ignore_size = False
3442

    
3443
  def Exec(self, feedback_fn):
3444
    """Activate the disks.
3445

3446
    """
3447
    disks_ok, disks_info = \
3448
              _AssembleInstanceDisks(self, self.instance,
3449
                                     ignore_size=self.op.ignore_size)
3450
    if not disks_ok:
3451
      raise errors.OpExecError("Cannot activate block devices")
3452

    
3453
    return disks_info
3454

    
3455

    
3456
def _AssembleInstanceDisks(lu, instance, ignore_secondaries=False,
3457
                           ignore_size=False):
3458
  """Prepare the block devices for an instance.
3459

3460
  This sets up the block devices on all nodes.
3461

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

3477
  """
3478
  device_info = []
3479
  disks_ok = True
3480
  iname = instance.name
3481
  # With the two passes mechanism we try to reduce the window of
3482
  # opportunity for the race condition of switching DRBD to primary
3483
  # before handshaking occured, but we do not eliminate it
3484

    
3485
  # The proper fix would be to wait (with some limits) until the
3486
  # connection has been made and drbd transitions from WFConnection
3487
  # into any other network-connected state (Connected, SyncTarget,
3488
  # SyncSource, etc.)
3489

    
3490
  # 1st pass, assemble on all nodes in secondary mode
3491
  for inst_disk in instance.disks:
3492
    for node, node_disk in inst_disk.ComputeNodeTree(instance.primary_node):
3493
      if ignore_size:
3494
        node_disk = node_disk.Copy()
3495
        node_disk.UnsetSize()
3496
      lu.cfg.SetDiskID(node_disk, node)
3497
      result = lu.rpc.call_blockdev_assemble(node, node_disk, iname, False)
3498
      msg = result.fail_msg
3499
      if msg:
3500
        lu.proc.LogWarning("Could not prepare block device %s on node %s"
3501
                           " (is_primary=False, pass=1): %s",
3502
                           inst_disk.iv_name, node, msg)
3503
        if not ignore_secondaries:
3504
          disks_ok = False
3505

    
3506
  # FIXME: race condition on drbd migration to primary
3507

    
3508
  # 2nd pass, do only the primary node
3509
  for inst_disk in instance.disks:
3510
    dev_path = None
3511

    
3512
    for node, node_disk in inst_disk.ComputeNodeTree(instance.primary_node):
3513
      if node != instance.primary_node:
3514
        continue
3515
      if ignore_size:
3516
        node_disk = node_disk.Copy()
3517
        node_disk.UnsetSize()
3518
      lu.cfg.SetDiskID(node_disk, node)
3519
      result = lu.rpc.call_blockdev_assemble(node, node_disk, iname, True)
3520
      msg = result.fail_msg
3521
      if msg:
3522
        lu.proc.LogWarning("Could not prepare block device %s on node %s"
3523
                           " (is_primary=True, pass=2): %s",
3524
                           inst_disk.iv_name, node, msg)
3525
        disks_ok = False
3526
      else:
3527
        dev_path = result.payload
3528

    
3529
    device_info.append((instance.primary_node, inst_disk.iv_name, dev_path))
3530

    
3531
  # leave the disks configured for the primary node
3532
  # this is a workaround that would be fixed better by
3533
  # improving the logical/physical id handling
3534
  for disk in instance.disks:
3535
    lu.cfg.SetDiskID(disk, instance.primary_node)
3536

    
3537
  return disks_ok, device_info
3538

    
3539

    
3540
def _StartInstanceDisks(lu, instance, force):
3541
  """Start the disks of an instance.
3542

3543
  """
3544
  disks_ok, _ = _AssembleInstanceDisks(lu, instance,
3545
                                           ignore_secondaries=force)
3546
  if not disks_ok:
3547
    _ShutdownInstanceDisks(lu, instance)
3548
    if force is not None and not force:
3549
      lu.proc.LogWarning("", hint="If the message above refers to a"
3550
                         " secondary node,"
3551
                         " you can retry the operation using '--force'.")
3552
    raise errors.OpExecError("Disk consistency error")
3553

    
3554

    
3555
class LUDeactivateInstanceDisks(NoHooksLU):
3556
  """Shutdown an instance's disks.
3557

3558
  """
3559
  _OP_REQP = ["instance_name"]
3560
  REQ_BGL = False
3561

    
3562
  def ExpandNames(self):
3563
    self._ExpandAndLockInstance()
3564
    self.needed_locks[locking.LEVEL_NODE] = []
3565
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
3566

    
3567
  def DeclareLocks(self, level):
3568
    if level == locking.LEVEL_NODE:
3569
      self._LockInstancesNodes()
3570

    
3571
  def CheckPrereq(self):
3572
    """Check prerequisites.
3573

3574
    This checks that the instance is in the cluster.
3575

3576
    """
3577
    self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
3578
    assert self.instance is not None, \
3579
      "Cannot retrieve locked instance %s" % self.op.instance_name
3580

    
3581
  def Exec(self, feedback_fn):
3582
    """Deactivate the disks
3583

3584
    """
3585
    instance = self.instance
3586
    _SafeShutdownInstanceDisks(self, instance)
3587

    
3588

    
3589
def _SafeShutdownInstanceDisks(lu, instance):
3590
  """Shutdown block devices of an instance.
3591

3592
  This function checks if an instance is running, before calling
3593
  _ShutdownInstanceDisks.
3594

3595
  """
3596
  pnode = instance.primary_node
3597
  ins_l = lu.rpc.call_instance_list([pnode], [instance.hypervisor])[pnode]
3598
  ins_l.Raise("Can't contact node %s" % pnode)
3599

    
3600
  if instance.name in ins_l.payload:
3601
    raise errors.OpExecError("Instance is running, can't shutdown"
3602
                             " block devices.")
3603

    
3604
  _ShutdownInstanceDisks(lu, instance)
3605

    
3606

    
3607
def _ShutdownInstanceDisks(lu, instance, ignore_primary=False):
3608
  """Shutdown block devices of an instance.
3609

3610
  This does the shutdown on all nodes of the instance.
3611

3612
  If the ignore_primary is false, errors on the primary node are
3613
  ignored.
3614

3615
  """
3616
  all_result = True
3617
  for disk in instance.disks:
3618
    for node, top_disk in disk.ComputeNodeTree(instance.primary_node):
3619
      lu.cfg.SetDiskID(top_disk, node)
3620
      result = lu.rpc.call_blockdev_shutdown(node, top_disk)
3621
      msg = result.fail_msg
3622
      if msg:
3623
        lu.LogWarning("Could not shutdown block device %s on node %s: %s",
3624
                      disk.iv_name, node, msg)
3625
        if not ignore_primary or node != instance.primary_node:
3626
          all_result = False
3627
  return all_result
3628

    
3629

    
3630
def _CheckNodeFreeMemory(lu, node, reason, requested, hypervisor_name):
3631
  """Checks if a node has enough free memory.
3632

3633
  This function check if a given node has the needed amount of free
3634
  memory. In case the node has less memory or we cannot get the
3635
  information from the node, this function raise an OpPrereqError
3636
  exception.
3637

3638
  @type lu: C{LogicalUnit}
3639
  @param lu: a logical unit from which we get configuration data
3640
  @type node: C{str}
3641
  @param node: the node to check
3642
  @type reason: C{str}
3643
  @param reason: string to use in the error message
3644
  @type requested: C{int}
3645
  @param requested: the amount of memory in MiB to check for
3646
  @type hypervisor_name: C{str}
3647
  @param hypervisor_name: the hypervisor to ask for memory stats
3648
  @raise errors.OpPrereqError: if the node doesn't have enough memory, or
3649
      we cannot check the node
3650

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

    
3666

    
3667
class LUStartupInstance(LogicalUnit):
3668
  """Starts an instance.
3669

3670
  """
3671
  HPATH = "instance-start"
3672
  HTYPE = constants.HTYPE_INSTANCE
3673
  _OP_REQP = ["instance_name", "force"]
3674
  REQ_BGL = False
3675

    
3676
  def ExpandNames(self):
3677
    self._ExpandAndLockInstance()
3678

    
3679
  def BuildHooksEnv(self):
3680
    """Build hooks env.
3681

3682
    This runs on master, primary and secondary nodes of the instance.
3683

3684
    """
3685
    env = {
3686
      "FORCE": self.op.force,
3687
      }
3688
    env.update(_BuildInstanceHookEnvByObject(self, self.instance))
3689
    nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
3690
    return env, nl, nl
3691

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

3695
    This checks that the instance is in the cluster.
3696

3697
    """
3698
    self.instance = instance = self.cfg.GetInstanceInfo(self.op.instance_name)
3699
    assert self.instance is not None, \
3700
      "Cannot retrieve locked instance %s" % self.op.instance_name
3701

    
3702
    # extra beparams
3703
    self.beparams = getattr(self.op, "beparams", {})
3704
    if self.beparams:
3705
      if not isinstance(self.beparams, dict):
3706
        raise errors.OpPrereqError("Invalid beparams passed: %s, expected"
3707
                                   " dict" % (type(self.beparams), ),
3708
                                   errors.ECODE_INVAL)
3709
      # fill the beparams dict
3710
      utils.ForceDictType(self.beparams, constants.BES_PARAMETER_TYPES)
3711
      self.op.beparams = self.beparams
3712

    
3713
    # extra hvparams
3714
    self.hvparams = getattr(self.op, "hvparams", {})
3715
    if self.hvparams:
3716
      if not isinstance(self.hvparams, dict):
3717
        raise errors.OpPrereqError("Invalid hvparams passed: %s, expected"
3718
                                   " dict" % (type(self.hvparams), ),
3719
                                   errors.ECODE_INVAL)
3720

    
3721
      # check hypervisor parameter syntax (locally)
3722
      cluster = self.cfg.GetClusterInfo()
3723
      utils.ForceDictType(self.hvparams, constants.HVS_PARAMETER_TYPES)
3724
      filled_hvp = objects.FillDict(cluster.hvparams[instance.hypervisor],
3725
                                    instance.hvparams)
3726
      filled_hvp.update(self.hvparams)
3727
      hv_type = hypervisor.GetHypervisor(instance.hypervisor)
3728
      hv_type.CheckParameterSyntax(filled_hvp)
3729
      _CheckHVParams(self, instance.all_nodes, instance.hypervisor, filled_hvp)
3730
      self.op.hvparams = self.hvparams
3731

    
3732
    _CheckNodeOnline(self, instance.primary_node)
3733

    
3734
    bep = self.cfg.GetClusterInfo().FillBE(instance)
3735
    # check bridges existence
3736
    _CheckInstanceBridgesExist(self, instance)
3737

    
3738
    remote_info = self.rpc.call_instance_info(instance.primary_node,
3739
                                              instance.name,
3740
                                              instance.hypervisor)
3741
    remote_info.Raise("Error checking node %s" % instance.primary_node,
3742
                      prereq=True, ecode=errors.ECODE_ENVIRON)
3743
    if not remote_info.payload: # not running already
3744
      _CheckNodeFreeMemory(self, instance.primary_node,
3745
                           "starting instance %s" % instance.name,
3746
                           bep[constants.BE_MEMORY], instance.hypervisor)
3747

    
3748
  def Exec(self, feedback_fn):
3749
    """Start the instance.
3750

3751
    """
3752
    instance = self.instance
3753
    force = self.op.force
3754

    
3755
    self.cfg.MarkInstanceUp(instance.name)
3756

    
3757
    node_current = instance.primary_node
3758

    
3759
    _StartInstanceDisks(self, instance, force)
3760

    
3761
    result = self.rpc.call_instance_start(node_current, instance,
3762
                                          self.hvparams, self.beparams)
3763
    msg = result.fail_msg
3764
    if msg:
3765
      _ShutdownInstanceDisks(self, instance)
3766
      raise errors.OpExecError("Could not start instance: %s" % msg)
3767

    
3768

    
3769
class LURebootInstance(LogicalUnit):
3770
  """Reboot an instance.
3771

3772
  """
3773
  HPATH = "instance-reboot"
3774
  HTYPE = constants.HTYPE_INSTANCE
3775
  _OP_REQP = ["instance_name", "ignore_secondaries", "reboot_type"]
3776
  REQ_BGL = False
3777

    
3778
  def CheckArguments(self):
3779
    """Check the arguments.
3780

3781
    """
3782
    self.shutdown_timeout = getattr(self.op, "shutdown_timeout",
3783
                                    constants.DEFAULT_SHUTDOWN_TIMEOUT)
3784

    
3785
  def ExpandNames(self):
3786
    if self.op.reboot_type not in [constants.INSTANCE_REBOOT_SOFT,
3787
                                   constants.INSTANCE_REBOOT_HARD,
3788
                                   constants.INSTANCE_REBOOT_FULL]:
3789
      raise errors.ParameterError("reboot type not in [%s, %s, %s]" %
3790
                                  (constants.INSTANCE_REBOOT_SOFT,
3791
                                   constants.INSTANCE_REBOOT_HARD,
3792
                                   constants.INSTANCE_REBOOT_FULL))
3793
    self._ExpandAndLockInstance()
3794

    
3795
  def BuildHooksEnv(self):
3796
    """Build hooks env.
3797

3798
    This runs on master, primary and secondary nodes of the instance.
3799

3800
    """
3801
    env = {
3802
      "IGNORE_SECONDARIES": self.op.ignore_secondaries,
3803
      "REBOOT_TYPE": self.op.reboot_type,
3804
      "SHUTDOWN_TIMEOUT": self.shutdown_timeout,
3805
      }
3806
    env.update(_BuildInstanceHookEnvByObject(self, self.instance))
3807
    nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
3808
    return env, nl, nl
3809

    
3810
  def CheckPrereq(self):
3811
    """Check prerequisites.
3812

3813
    This checks that the instance is in the cluster.
3814

3815
    """
3816
    self.instance = instance = self.cfg.GetInstanceInfo(self.op.instance_name)
3817
    assert self.instance is not None, \
3818
      "Cannot retrieve locked instance %s" % self.op.instance_name
3819

    
3820
    _CheckNodeOnline(self, instance.primary_node)
3821

    
3822
    # check bridges existence
3823
    _CheckInstanceBridgesExist(self, instance)
3824

    
3825
  def Exec(self, feedback_fn):
3826
    """Reboot the instance.
3827

3828
    """
3829
    instance = self.instance
3830
    ignore_secondaries = self.op.ignore_secondaries
3831
    reboot_type = self.op.reboot_type
3832

    
3833
    node_current = instance.primary_node
3834

    
3835
    if reboot_type in [constants.INSTANCE_REBOOT_SOFT,
3836
                       constants.INSTANCE_REBOOT_HARD]:
3837
      for disk in instance.disks:
3838
        self.cfg.SetDiskID(disk, node_current)
3839
      result = self.rpc.call_instance_reboot(node_current, instance,
3840
                                             reboot_type,
3841
                                             self.shutdown_timeout)
3842
      result.Raise("Could not reboot instance")
3843
    else:
3844
      result = self.rpc.call_instance_shutdown(node_current, instance,
3845
                                               self.shutdown_timeout)
3846
      result.Raise("Could not shutdown instance for full reboot")
3847
      _ShutdownInstanceDisks(self, instance)
3848
      _StartInstanceDisks(self, instance, ignore_secondaries)
3849
      result = self.rpc.call_instance_start(node_current, instance, None, None)
3850
      msg = result.fail_msg
3851
      if msg:
3852
        _ShutdownInstanceDisks(self, instance)
3853
        raise errors.OpExecError("Could not start instance for"
3854
                                 " full reboot: %s" % msg)
3855

    
3856
    self.cfg.MarkInstanceUp(instance.name)
3857

    
3858

    
3859
class LUShutdownInstance(LogicalUnit):
3860
  """Shutdown an instance.
3861

3862
  """
3863
  HPATH = "instance-stop"
3864
  HTYPE = constants.HTYPE_INSTANCE
3865
  _OP_REQP = ["instance_name"]
3866
  REQ_BGL = False
3867

    
3868
  def CheckArguments(self):
3869
    """Check the arguments.
3870

3871
    """
3872
    self.timeout = getattr(self.op, "timeout",
3873
                           constants.DEFAULT_SHUTDOWN_TIMEOUT)
3874

    
3875
  def ExpandNames(self):
3876
    self._ExpandAndLockInstance()
3877

    
3878
  def BuildHooksEnv(self):
3879
    """Build hooks env.
3880

3881
    This runs on master, primary and secondary nodes of the instance.
3882

3883
    """
3884
    env = _BuildInstanceHookEnvByObject(self, self.instance)
3885
    env["TIMEOUT"] = self.timeout
3886
    nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
3887
    return env, nl, nl
3888

    
3889
  def CheckPrereq(self):
3890
    """Check prerequisites.
3891

3892
    This checks that the instance is in the cluster.
3893

3894
    """
3895
    self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
3896
    assert self.instance is not None, \
3897
      "Cannot retrieve locked instance %s" % self.op.instance_name
3898
    _CheckNodeOnline(self, self.instance.primary_node)
3899

    
3900
  def Exec(self, feedback_fn):
3901
    """Shutdown the instance.
3902

3903
    """
3904
    instance = self.instance
3905
    node_current = instance.primary_node
3906
    timeout = self.timeout
3907
    self.cfg.MarkInstanceDown(instance.name)
3908
    result = self.rpc.call_instance_shutdown(node_current, instance, timeout)
3909
    msg = result.fail_msg
3910
    if msg:
3911
      self.proc.LogWarning("Could not shutdown instance: %s" % msg)
3912

    
3913
    _ShutdownInstanceDisks(self, instance)
3914

    
3915

    
3916
class LUReinstallInstance(LogicalUnit):
3917
  """Reinstall an instance.
3918

3919
  """
3920
  HPATH = "instance-reinstall"
3921
  HTYPE = constants.HTYPE_INSTANCE
3922
  _OP_REQP = ["instance_name"]
3923
  REQ_BGL = False
3924

    
3925
  def ExpandNames(self):
3926
    self._ExpandAndLockInstance()
3927

    
3928
  def BuildHooksEnv(self):
3929
    """Build hooks env.
3930

3931
    This runs on master, primary and secondary nodes of the instance.
3932

3933
    """
3934
    env = _BuildInstanceHookEnvByObject(self, self.instance)
3935
    nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
3936
    return env, nl, nl
3937

    
3938
  def CheckPrereq(self):
3939
    """Check prerequisites.
3940

3941
    This checks that the instance is in the cluster and is not running.
3942

3943
    """
3944
    instance = self.cfg.GetInstanceInfo(self.op.instance_name)
3945
    assert instance is not None, \
3946
      "Cannot retrieve locked instance %s" % self.op.instance_name
3947
    _CheckNodeOnline(self, instance.primary_node)
3948

    
3949
    if instance.disk_template == constants.DT_DISKLESS:
3950
      raise errors.OpPrereqError("Instance '%s' has no disks" %
3951
                                 self.op.instance_name,
3952
                                 errors.ECODE_INVAL)
3953
    if instance.admin_up:
3954
      raise errors.OpPrereqError("Instance '%s' is marked to be up" %
3955
                                 self.op.instance_name,
3956
                                 errors.ECODE_STATE)
3957
    remote_info = self.rpc.call_instance_info(instance.primary_node,
3958
                                              instance.name,
3959
                                              instance.hypervisor)
3960
    remote_info.Raise("Error checking node %s" % instance.primary_node,
3961
                      prereq=True, ecode=errors.ECODE_ENVIRON)
3962
    if remote_info.payload:
3963
      raise errors.OpPrereqError("Instance '%s' is running on the node %s" %
3964
                                 (self.op.instance_name,
3965
                                  instance.primary_node),
3966
                                 errors.ECODE_STATE)
3967

    
3968
    self.op.os_type = getattr(self.op, "os_type", None)
3969
    self.op.force_variant = getattr(self.op, "force_variant", False)
3970
    if self.op.os_type is not None:
3971
      # OS verification
3972
      pnode = self.cfg.GetNodeInfo(
3973
        self.cfg.ExpandNodeName(instance.primary_node))
3974
      if pnode is None:
3975
        raise errors.OpPrereqError("Primary node '%s' is unknown" %
3976
                                   self.op.pnode, errors.ECODE_NOENT)
3977
      result = self.rpc.call_os_get(pnode.name, self.op.os_type)
3978
      result.Raise("OS '%s' not in supported OS list for primary node %s" %
3979
                   (self.op.os_type, pnode.name),
3980
                   prereq=True, ecode=errors.ECODE_INVAL)
3981
      if not self.op.force_variant:
3982
        _CheckOSVariant(result.payload, self.op.os_type)
3983

    
3984
    self.instance = instance
3985

    
3986
  def Exec(self, feedback_fn):
3987
    """Reinstall the instance.
3988

3989
    """
3990
    inst = self.instance
3991

    
3992
    if self.op.os_type is not None:
3993
      feedback_fn("Changing OS to '%s'..." % self.op.os_type)
3994
      inst.os = self.op.os_type
3995
      self.cfg.Update(inst, feedback_fn)
3996

    
3997
    _StartInstanceDisks(self, inst, None)
3998
    try:
3999
      feedback_fn("Running the instance OS create scripts...")
4000
      # FIXME: pass debug option from opcode to backend
4001
      result = self.rpc.call_instance_os_add(inst.primary_node, inst, True, 0)
4002
      result.Raise("Could not install OS for instance %s on node %s" %
4003
                   (inst.name, inst.primary_node))
4004
    finally:
4005
      _ShutdownInstanceDisks(self, inst)
4006

    
4007

    
4008
class LURecreateInstanceDisks(LogicalUnit):
4009
  """Recreate an instance's missing disks.