Statistics
| Branch: | Tag: | Revision:

root / lib / cmdlib / instance_storage.py @ 125e1230

History | View | Annotate | Download (93 kB)

1
#
2
#
3

    
4
# Copyright (C) 2006, 2007, 2008, 2009, 2010, 2011, 2012, 2013 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
"""Logical units dealing with storage of instances."""
23

    
24
import itertools
25
import logging
26
import os
27
import time
28

    
29
from ganeti import compat
30
from ganeti import constants
31
from ganeti import errors
32
from ganeti import ht
33
from ganeti import locking
34
from ganeti.masterd import iallocator
35
from ganeti import objects
36
from ganeti import utils
37
from ganeti import opcodes
38
from ganeti import rpc
39
from ganeti.cmdlib.base import LogicalUnit, NoHooksLU, Tasklet
40
from ganeti.cmdlib.common import INSTANCE_DOWN, INSTANCE_NOT_RUNNING, \
41
  AnnotateDiskParams, CheckIAllocatorOrNode, ExpandNodeName, \
42
  CheckNodeOnline, CheckInstanceNodeGroups, CheckInstanceState, \
43
  IsExclusiveStorageEnabledNode, FindFaultyInstanceDisks
44
from ganeti.cmdlib.instance_utils import GetInstanceInfoText, \
45
  CopyLockList, ReleaseLocks, CheckNodeVmCapable, \
46
  BuildInstanceHookEnvByObject, CheckNodeNotDrained, CheckTargetNodeIPolicy
47

    
48
import ganeti.masterd.instance
49

    
50

    
51
_DISK_TEMPLATE_NAME_PREFIX = {
52
  constants.DT_PLAIN: "",
53
  constants.DT_RBD: ".rbd",
54
  constants.DT_EXT: ".ext",
55
  }
56

    
57

    
58
_DISK_TEMPLATE_DEVICE_TYPE = {
59
  constants.DT_PLAIN: constants.LD_LV,
60
  constants.DT_FILE: constants.LD_FILE,
61
  constants.DT_SHARED_FILE: constants.LD_FILE,
62
  constants.DT_BLOCK: constants.LD_BLOCKDEV,
63
  constants.DT_RBD: constants.LD_RBD,
64
  constants.DT_EXT: constants.LD_EXT,
65
  }
66

    
67

    
68
def CreateSingleBlockDev(lu, node, instance, device, info, force_open,
69
                         excl_stor):
70
  """Create a single block device on a given node.
71

72
  This will not recurse over children of the device, so they must be
73
  created in advance.
74

75
  @param lu: the lu on whose behalf we execute
76
  @param node: the node on which to create the device
77
  @type instance: L{objects.Instance}
78
  @param instance: the instance which owns the device
79
  @type device: L{objects.Disk}
80
  @param device: the device to create
81
  @param info: the extra 'metadata' we should attach to the device
82
      (this will be represented as a LVM tag)
83
  @type force_open: boolean
84
  @param force_open: this parameter will be passes to the
85
      L{backend.BlockdevCreate} function where it specifies
86
      whether we run on primary or not, and it affects both
87
      the child assembly and the device own Open() execution
88
  @type excl_stor: boolean
89
  @param excl_stor: Whether exclusive_storage is active for the node
90

91
  """
92
  lu.cfg.SetDiskID(device, node)
93
  result = lu.rpc.call_blockdev_create(node, device, device.size,
94
                                       instance.name, force_open, info,
95
                                       excl_stor)
96
  result.Raise("Can't create block device %s on"
97
               " node %s for instance %s" % (device, node, instance.name))
98
  if device.physical_id is None:
99
    device.physical_id = result.payload
100

    
101

    
102
def _CreateBlockDevInner(lu, node, instance, device, force_create,
103
                         info, force_open, excl_stor):
104
  """Create a tree of block devices on a given node.
105

106
  If this device type has to be created on secondaries, create it and
107
  all its children.
108

109
  If not, just recurse to children keeping the same 'force' value.
110

111
  @attention: The device has to be annotated already.
112

113
  @param lu: the lu on whose behalf we execute
114
  @param node: the node on which to create the device
115
  @type instance: L{objects.Instance}
116
  @param instance: the instance which owns the device
117
  @type device: L{objects.Disk}
118
  @param device: the device to create
119
  @type force_create: boolean
120
  @param force_create: whether to force creation of this device; this
121
      will be change to True whenever we find a device which has
122
      CreateOnSecondary() attribute
123
  @param info: the extra 'metadata' we should attach to the device
124
      (this will be represented as a LVM tag)
125
  @type force_open: boolean
126
  @param force_open: this parameter will be passes to the
127
      L{backend.BlockdevCreate} function where it specifies
128
      whether we run on primary or not, and it affects both
129
      the child assembly and the device own Open() execution
130
  @type excl_stor: boolean
131
  @param excl_stor: Whether exclusive_storage is active for the node
132

133
  @return: list of created devices
134
  """
135
  created_devices = []
136
  try:
137
    if device.CreateOnSecondary():
138
      force_create = True
139

    
140
    if device.children:
141
      for child in device.children:
142
        devs = _CreateBlockDevInner(lu, node, instance, child, force_create,
143
                                    info, force_open, excl_stor)
144
        created_devices.extend(devs)
145

    
146
    if not force_create:
147
      return created_devices
148

    
149
    CreateSingleBlockDev(lu, node, instance, device, info, force_open,
150
                         excl_stor)
151
    # The device has been completely created, so there is no point in keeping
152
    # its subdevices in the list. We just add the device itself instead.
153
    created_devices = [(node, device)]
154
    return created_devices
155

    
156
  except errors.DeviceCreationError, e:
157
    e.created_devices.extend(created_devices)
158
    raise e
159
  except errors.OpExecError, e:
160
    raise errors.DeviceCreationError(str(e), created_devices)
161

    
162

    
163
def IsExclusiveStorageEnabledNodeName(cfg, nodename):
164
  """Whether exclusive_storage is in effect for the given node.
165

166
  @type cfg: L{config.ConfigWriter}
167
  @param cfg: The cluster configuration
168
  @type nodename: string
169
  @param nodename: The node
170
  @rtype: bool
171
  @return: The effective value of exclusive_storage
172
  @raise errors.OpPrereqError: if no node exists with the given name
173

174
  """
175
  ni = cfg.GetNodeInfo(nodename)
176
  if ni is None:
177
    raise errors.OpPrereqError("Invalid node name %s" % nodename,
178
                               errors.ECODE_NOENT)
179
  return IsExclusiveStorageEnabledNode(cfg, ni)
180

    
181

    
182
def _CreateBlockDev(lu, node, instance, device, force_create, info,
183
                    force_open):
184
  """Wrapper around L{_CreateBlockDevInner}.
185

186
  This method annotates the root device first.
187

188
  """
189
  (disk,) = AnnotateDiskParams(instance, [device], lu.cfg)
190
  excl_stor = IsExclusiveStorageEnabledNodeName(lu.cfg, node)
191
  return _CreateBlockDevInner(lu, node, instance, disk, force_create, info,
192
                              force_open, excl_stor)
193

    
194

    
195
def _UndoCreateDisks(lu, disks_created):
196
  """Undo the work performed by L{CreateDisks}.
197

198
  This function is called in case of an error to undo the work of
199
  L{CreateDisks}.
200

201
  @type lu: L{LogicalUnit}
202
  @param lu: the logical unit on whose behalf we execute
203
  @param disks_created: the result returned by L{CreateDisks}
204

205
  """
206
  for (node, disk) in disks_created:
207
    lu.cfg.SetDiskID(disk, node)
208
    result = lu.rpc.call_blockdev_remove(node, disk)
209
    if result.fail_msg:
210
      logging.warning("Failed to remove newly-created disk %s on node %s:"
211
                      " %s", disk, node, result.fail_msg)
212

    
213

    
214
def CreateDisks(lu, instance, to_skip=None, target_node=None, disks=None):
215
  """Create all disks for an instance.
216

217
  This abstracts away some work from AddInstance.
218

219
  @type lu: L{LogicalUnit}
220
  @param lu: the logical unit on whose behalf we execute
221
  @type instance: L{objects.Instance}
222
  @param instance: the instance whose disks we should create
223
  @type to_skip: list
224
  @param to_skip: list of indices to skip
225
  @type target_node: string
226
  @param target_node: if passed, overrides the target node for creation
227
  @type disks: list of {objects.Disk}
228
  @param disks: the disks to create; if not specified, all the disks of the
229
      instance are created
230
  @return: information about the created disks, to be used to call
231
      L{_UndoCreateDisks}
232
  @raise errors.OpPrereqError: in case of error
233

234
  """
235
  info = GetInstanceInfoText(instance)
236
  if target_node is None:
237
    pnode = instance.primary_node
238
    all_nodes = instance.all_nodes
239
  else:
240
    pnode = target_node
241
    all_nodes = [pnode]
242

    
243
  if disks is None:
244
    disks = instance.disks
245

    
246
  if instance.disk_template in constants.DTS_FILEBASED:
247
    file_storage_dir = os.path.dirname(instance.disks[0].logical_id[1])
248
    result = lu.rpc.call_file_storage_dir_create(pnode, file_storage_dir)
249

    
250
    result.Raise("Failed to create directory '%s' on"
251
                 " node %s" % (file_storage_dir, pnode))
252

    
253
  disks_created = []
254
  for idx, device in enumerate(disks):
255
    if to_skip and idx in to_skip:
256
      continue
257
    logging.info("Creating disk %s for instance '%s'", idx, instance.name)
258
    for node in all_nodes:
259
      f_create = node == pnode
260
      try:
261
        _CreateBlockDev(lu, node, instance, device, f_create, info, f_create)
262
        disks_created.append((node, device))
263
      except errors.DeviceCreationError, e:
264
        logging.warning("Creating disk %s for instance '%s' failed",
265
                        idx, instance.name)
266
        disks_created.extend(e.created_devices)
267
        _UndoCreateDisks(lu, disks_created)
268
        raise errors.OpExecError(e.message)
269
  return disks_created
270

    
271

    
272
def ComputeDiskSizePerVG(disk_template, disks):
273
  """Compute disk size requirements in the volume group
274

275
  """
276
  def _compute(disks, payload):
277
    """Universal algorithm.
278

279
    """
280
    vgs = {}
281
    for disk in disks:
282
      vgs[disk[constants.IDISK_VG]] = \
283
        vgs.get(constants.IDISK_VG, 0) + disk[constants.IDISK_SIZE] + payload
284

    
285
    return vgs
286

    
287
  # Required free disk space as a function of disk and swap space
288
  req_size_dict = {
289
    constants.DT_DISKLESS: {},
290
    constants.DT_PLAIN: _compute(disks, 0),
291
    # 128 MB are added for drbd metadata for each disk
292
    constants.DT_DRBD8: _compute(disks, constants.DRBD_META_SIZE),
293
    constants.DT_FILE: {},
294
    constants.DT_SHARED_FILE: {},
295
    }
296

    
297
  if disk_template not in req_size_dict:
298
    raise errors.ProgrammerError("Disk template '%s' size requirement"
299
                                 " is unknown" % disk_template)
300

    
301
  return req_size_dict[disk_template]
302

    
303

    
304
def ComputeDisks(op, default_vg):
305
  """Computes the instance disks.
306

307
  @param op: The instance opcode
308
  @param default_vg: The default_vg to assume
309

310
  @return: The computed disks
311

312
  """
313
  disks = []
314
  for disk in op.disks:
315
    mode = disk.get(constants.IDISK_MODE, constants.DISK_RDWR)
316
    if mode not in constants.DISK_ACCESS_SET:
317
      raise errors.OpPrereqError("Invalid disk access mode '%s'" %
318
                                 mode, errors.ECODE_INVAL)
319
    size = disk.get(constants.IDISK_SIZE, None)
320
    if size is None:
321
      raise errors.OpPrereqError("Missing disk size", errors.ECODE_INVAL)
322
    try:
323
      size = int(size)
324
    except (TypeError, ValueError):
325
      raise errors.OpPrereqError("Invalid disk size '%s'" % size,
326
                                 errors.ECODE_INVAL)
327

    
328
    ext_provider = disk.get(constants.IDISK_PROVIDER, None)
329
    if ext_provider and op.disk_template != constants.DT_EXT:
330
      raise errors.OpPrereqError("The '%s' option is only valid for the %s"
331
                                 " disk template, not %s" %
332
                                 (constants.IDISK_PROVIDER, constants.DT_EXT,
333
                                  op.disk_template), errors.ECODE_INVAL)
334

    
335
    data_vg = disk.get(constants.IDISK_VG, default_vg)
336
    name = disk.get(constants.IDISK_NAME, None)
337
    if name is not None and name.lower() == constants.VALUE_NONE:
338
      name = None
339
    new_disk = {
340
      constants.IDISK_SIZE: size,
341
      constants.IDISK_MODE: mode,
342
      constants.IDISK_VG: data_vg,
343
      constants.IDISK_NAME: name,
344
      }
345

    
346
    if constants.IDISK_METAVG in disk:
347
      new_disk[constants.IDISK_METAVG] = disk[constants.IDISK_METAVG]
348
    if constants.IDISK_ADOPT in disk:
349
      new_disk[constants.IDISK_ADOPT] = disk[constants.IDISK_ADOPT]
350

    
351
    # For extstorage, demand the `provider' option and add any
352
    # additional parameters (ext-params) to the dict
353
    if op.disk_template == constants.DT_EXT:
354
      if ext_provider:
355
        new_disk[constants.IDISK_PROVIDER] = ext_provider
356
        for key in disk:
357
          if key not in constants.IDISK_PARAMS:
358
            new_disk[key] = disk[key]
359
      else:
360
        raise errors.OpPrereqError("Missing provider for template '%s'" %
361
                                   constants.DT_EXT, errors.ECODE_INVAL)
362

    
363
    disks.append(new_disk)
364

    
365
  return disks
366

    
367

    
368
def CheckRADOSFreeSpace():
369
  """Compute disk size requirements inside the RADOS cluster.
370

371
  """
372
  # For the RADOS cluster we assume there is always enough space.
373
  pass
374

    
375

    
376
def _GenerateDRBD8Branch(lu, primary, secondary, size, vgnames, names,
377
                         iv_name, p_minor, s_minor):
378
  """Generate a drbd8 device complete with its children.
379

380
  """
381
  assert len(vgnames) == len(names) == 2
382
  port = lu.cfg.AllocatePort()
383
  shared_secret = lu.cfg.GenerateDRBDSecret(lu.proc.GetECId())
384

    
385
  dev_data = objects.Disk(dev_type=constants.LD_LV, size=size,
386
                          logical_id=(vgnames[0], names[0]),
387
                          params={})
388
  dev_data.uuid = lu.cfg.GenerateUniqueID(lu.proc.GetECId())
389
  dev_meta = objects.Disk(dev_type=constants.LD_LV,
390
                          size=constants.DRBD_META_SIZE,
391
                          logical_id=(vgnames[1], names[1]),
392
                          params={})
393
  dev_meta.uuid = lu.cfg.GenerateUniqueID(lu.proc.GetECId())
394
  drbd_dev = objects.Disk(dev_type=constants.LD_DRBD8, size=size,
395
                          logical_id=(primary, secondary, port,
396
                                      p_minor, s_minor,
397
                                      shared_secret),
398
                          children=[dev_data, dev_meta],
399
                          iv_name=iv_name, params={})
400
  drbd_dev.uuid = lu.cfg.GenerateUniqueID(lu.proc.GetECId())
401
  return drbd_dev
402

    
403

    
404
def GenerateDiskTemplate(
405
  lu, template_name, instance_name, primary_node, secondary_nodes,
406
  disk_info, file_storage_dir, file_driver, base_index,
407
  feedback_fn, full_disk_params, _req_file_storage=opcodes.RequireFileStorage,
408
  _req_shr_file_storage=opcodes.RequireSharedFileStorage):
409
  """Generate the entire disk layout for a given template type.
410

411
  """
412
  vgname = lu.cfg.GetVGName()
413
  disk_count = len(disk_info)
414
  disks = []
415

    
416
  if template_name == constants.DT_DISKLESS:
417
    pass
418
  elif template_name == constants.DT_DRBD8:
419
    if len(secondary_nodes) != 1:
420
      raise errors.ProgrammerError("Wrong template configuration")
421
    remote_node = secondary_nodes[0]
422
    minors = lu.cfg.AllocateDRBDMinor(
423
      [primary_node, remote_node] * len(disk_info), instance_name)
424

    
425
    (drbd_params, _, _) = objects.Disk.ComputeLDParams(template_name,
426
                                                       full_disk_params)
427
    drbd_default_metavg = drbd_params[constants.LDP_DEFAULT_METAVG]
428

    
429
    names = []
430
    for lv_prefix in _GenerateUniqueNames(lu, [".disk%d" % (base_index + i)
431
                                               for i in range(disk_count)]):
432
      names.append(lv_prefix + "_data")
433
      names.append(lv_prefix + "_meta")
434
    for idx, disk in enumerate(disk_info):
435
      disk_index = idx + base_index
436
      data_vg = disk.get(constants.IDISK_VG, vgname)
437
      meta_vg = disk.get(constants.IDISK_METAVG, drbd_default_metavg)
438
      disk_dev = _GenerateDRBD8Branch(lu, primary_node, remote_node,
439
                                      disk[constants.IDISK_SIZE],
440
                                      [data_vg, meta_vg],
441
                                      names[idx * 2:idx * 2 + 2],
442
                                      "disk/%d" % disk_index,
443
                                      minors[idx * 2], minors[idx * 2 + 1])
444
      disk_dev.mode = disk[constants.IDISK_MODE]
445
      disk_dev.name = disk.get(constants.IDISK_NAME, None)
446
      disks.append(disk_dev)
447
  else:
448
    if secondary_nodes:
449
      raise errors.ProgrammerError("Wrong template configuration")
450

    
451
    if template_name == constants.DT_FILE:
452
      _req_file_storage()
453
    elif template_name == constants.DT_SHARED_FILE:
454
      _req_shr_file_storage()
455

    
456
    name_prefix = _DISK_TEMPLATE_NAME_PREFIX.get(template_name, None)
457
    if name_prefix is None:
458
      names = None
459
    else:
460
      names = _GenerateUniqueNames(lu, ["%s.disk%s" %
461
                                        (name_prefix, base_index + i)
462
                                        for i in range(disk_count)])
463

    
464
    if template_name == constants.DT_PLAIN:
465

    
466
      def logical_id_fn(idx, _, disk):
467
        vg = disk.get(constants.IDISK_VG, vgname)
468
        return (vg, names[idx])
469

    
470
    elif template_name in (constants.DT_FILE, constants.DT_SHARED_FILE):
471
      logical_id_fn = \
472
        lambda _, disk_index, disk: (file_driver,
473
                                     "%s/disk%d" % (file_storage_dir,
474
                                                    disk_index))
475
    elif template_name == constants.DT_BLOCK:
476
      logical_id_fn = \
477
        lambda idx, disk_index, disk: (constants.BLOCKDEV_DRIVER_MANUAL,
478
                                       disk[constants.IDISK_ADOPT])
479
    elif template_name == constants.DT_RBD:
480
      logical_id_fn = lambda idx, _, disk: ("rbd", names[idx])
481
    elif template_name == constants.DT_EXT:
482
      def logical_id_fn(idx, _, disk):
483
        provider = disk.get(constants.IDISK_PROVIDER, None)
484
        if provider is None:
485
          raise errors.ProgrammerError("Disk template is %s, but '%s' is"
486
                                       " not found", constants.DT_EXT,
487
                                       constants.IDISK_PROVIDER)
488
        return (provider, names[idx])
489
    else:
490
      raise errors.ProgrammerError("Unknown disk template '%s'" % template_name)
491

    
492
    dev_type = _DISK_TEMPLATE_DEVICE_TYPE[template_name]
493

    
494
    for idx, disk in enumerate(disk_info):
495
      params = {}
496
      # Only for the Ext template add disk_info to params
497
      if template_name == constants.DT_EXT:
498
        params[constants.IDISK_PROVIDER] = disk[constants.IDISK_PROVIDER]
499
        for key in disk:
500
          if key not in constants.IDISK_PARAMS:
501
            params[key] = disk[key]
502
      disk_index = idx + base_index
503
      size = disk[constants.IDISK_SIZE]
504
      feedback_fn("* disk %s, size %s" %
505
                  (disk_index, utils.FormatUnit(size, "h")))
506
      disk_dev = objects.Disk(dev_type=dev_type, size=size,
507
                              logical_id=logical_id_fn(idx, disk_index, disk),
508
                              iv_name="disk/%d" % disk_index,
509
                              mode=disk[constants.IDISK_MODE],
510
                              params=params)
511
      disk_dev.name = disk.get(constants.IDISK_NAME, None)
512
      disk_dev.uuid = lu.cfg.GenerateUniqueID(lu.proc.GetECId())
513
      disks.append(disk_dev)
514

    
515
  return disks
516

    
517

    
518
class LUInstanceRecreateDisks(LogicalUnit):
519
  """Recreate an instance's missing disks.
520

521
  """
522
  HPATH = "instance-recreate-disks"
523
  HTYPE = constants.HTYPE_INSTANCE
524
  REQ_BGL = False
525

    
526
  _MODIFYABLE = compat.UniqueFrozenset([
527
    constants.IDISK_SIZE,
528
    constants.IDISK_MODE,
529
    ])
530

    
531
  # New or changed disk parameters may have different semantics
532
  assert constants.IDISK_PARAMS == (_MODIFYABLE | frozenset([
533
    constants.IDISK_ADOPT,
534

    
535
    # TODO: Implement support changing VG while recreating
536
    constants.IDISK_VG,
537
    constants.IDISK_METAVG,
538
    constants.IDISK_PROVIDER,
539
    constants.IDISK_NAME,
540
    ]))
541

    
542
  def _RunAllocator(self):
543
    """Run the allocator based on input opcode.
544

545
    """
546
    be_full = self.cfg.GetClusterInfo().FillBE(self.instance)
547

    
548
    # FIXME
549
    # The allocator should actually run in "relocate" mode, but current
550
    # allocators don't support relocating all the nodes of an instance at
551
    # the same time. As a workaround we use "allocate" mode, but this is
552
    # suboptimal for two reasons:
553
    # - The instance name passed to the allocator is present in the list of
554
    #   existing instances, so there could be a conflict within the
555
    #   internal structures of the allocator. This doesn't happen with the
556
    #   current allocators, but it's a liability.
557
    # - The allocator counts the resources used by the instance twice: once
558
    #   because the instance exists already, and once because it tries to
559
    #   allocate a new instance.
560
    # The allocator could choose some of the nodes on which the instance is
561
    # running, but that's not a problem. If the instance nodes are broken,
562
    # they should be already be marked as drained or offline, and hence
563
    # skipped by the allocator. If instance disks have been lost for other
564
    # reasons, then recreating the disks on the same nodes should be fine.
565
    disk_template = self.instance.disk_template
566
    spindle_use = be_full[constants.BE_SPINDLE_USE]
567
    req = iallocator.IAReqInstanceAlloc(name=self.op.instance_name,
568
                                        disk_template=disk_template,
569
                                        tags=list(self.instance.GetTags()),
570
                                        os=self.instance.os,
571
                                        nics=[{}],
572
                                        vcpus=be_full[constants.BE_VCPUS],
573
                                        memory=be_full[constants.BE_MAXMEM],
574
                                        spindle_use=spindle_use,
575
                                        disks=[{constants.IDISK_SIZE: d.size,
576
                                                constants.IDISK_MODE: d.mode}
577
                                               for d in self.instance.disks],
578
                                        hypervisor=self.instance.hypervisor,
579
                                        node_whitelist=None)
580
    ial = iallocator.IAllocator(self.cfg, self.rpc, req)
581

    
582
    ial.Run(self.op.iallocator)
583

    
584
    assert req.RequiredNodes() == len(self.instance.all_nodes)
585

    
586
    if not ial.success:
587
      raise errors.OpPrereqError("Can't compute nodes using iallocator '%s':"
588
                                 " %s" % (self.op.iallocator, ial.info),
589
                                 errors.ECODE_NORES)
590

    
591
    self.op.nodes = ial.result
592
    self.LogInfo("Selected nodes for instance %s via iallocator %s: %s",
593
                 self.op.instance_name, self.op.iallocator,
594
                 utils.CommaJoin(ial.result))
595

    
596
  def CheckArguments(self):
597
    if self.op.disks and ht.TNonNegativeInt(self.op.disks[0]):
598
      # Normalize and convert deprecated list of disk indices
599
      self.op.disks = [(idx, {}) for idx in sorted(frozenset(self.op.disks))]
600

    
601
    duplicates = utils.FindDuplicates(map(compat.fst, self.op.disks))
602
    if duplicates:
603
      raise errors.OpPrereqError("Some disks have been specified more than"
604
                                 " once: %s" % utils.CommaJoin(duplicates),
605
                                 errors.ECODE_INVAL)
606

    
607
    # We don't want _CheckIAllocatorOrNode selecting the default iallocator
608
    # when neither iallocator nor nodes are specified
609
    if self.op.iallocator or self.op.nodes:
610
      CheckIAllocatorOrNode(self, "iallocator", "nodes")
611

    
612
    for (idx, params) in self.op.disks:
613
      utils.ForceDictType(params, constants.IDISK_PARAMS_TYPES)
614
      unsupported = frozenset(params.keys()) - self._MODIFYABLE
615
      if unsupported:
616
        raise errors.OpPrereqError("Parameters for disk %s try to change"
617
                                   " unmodifyable parameter(s): %s" %
618
                                   (idx, utils.CommaJoin(unsupported)),
619
                                   errors.ECODE_INVAL)
620

    
621
  def ExpandNames(self):
622
    self._ExpandAndLockInstance()
623
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_APPEND
624

    
625
    if self.op.nodes:
626
      self.op.nodes = [ExpandNodeName(self.cfg, n) for n in self.op.nodes]
627
      self.needed_locks[locking.LEVEL_NODE] = list(self.op.nodes)
628
    else:
629
      self.needed_locks[locking.LEVEL_NODE] = []
630
      if self.op.iallocator:
631
        # iallocator will select a new node in the same group
632
        self.needed_locks[locking.LEVEL_NODEGROUP] = []
633
        self.needed_locks[locking.LEVEL_NODE_ALLOC] = locking.ALL_SET
634

    
635
    self.needed_locks[locking.LEVEL_NODE_RES] = []
636

    
637
  def DeclareLocks(self, level):
638
    if level == locking.LEVEL_NODEGROUP:
639
      assert self.op.iallocator is not None
640
      assert not self.op.nodes
641
      assert not self.needed_locks[locking.LEVEL_NODEGROUP]
642
      self.share_locks[locking.LEVEL_NODEGROUP] = 1
643
      # Lock the primary group used by the instance optimistically; this
644
      # requires going via the node before it's locked, requiring
645
      # verification later on
646
      self.needed_locks[locking.LEVEL_NODEGROUP] = \
647
        self.cfg.GetInstanceNodeGroups(self.op.instance_name, primary_only=True)
648

    
649
    elif level == locking.LEVEL_NODE:
650
      # If an allocator is used, then we lock all the nodes in the current
651
      # instance group, as we don't know yet which ones will be selected;
652
      # if we replace the nodes without using an allocator, locks are
653
      # already declared in ExpandNames; otherwise, we need to lock all the
654
      # instance nodes for disk re-creation
655
      if self.op.iallocator:
656
        assert not self.op.nodes
657
        assert not self.needed_locks[locking.LEVEL_NODE]
658
        assert len(self.owned_locks(locking.LEVEL_NODEGROUP)) == 1
659

    
660
        # Lock member nodes of the group of the primary node
661
        for group_uuid in self.owned_locks(locking.LEVEL_NODEGROUP):
662
          self.needed_locks[locking.LEVEL_NODE].extend(
663
            self.cfg.GetNodeGroup(group_uuid).members)
664

    
665
        assert locking.NAL in self.owned_locks(locking.LEVEL_NODE_ALLOC)
666
      elif not self.op.nodes:
667
        self._LockInstancesNodes(primary_only=False)
668
    elif level == locking.LEVEL_NODE_RES:
669
      # Copy node locks
670
      self.needed_locks[locking.LEVEL_NODE_RES] = \
671
        CopyLockList(self.needed_locks[locking.LEVEL_NODE])
672

    
673
  def BuildHooksEnv(self):
674
    """Build hooks env.
675

676
    This runs on master, primary and secondary nodes of the instance.
677

678
    """
679
    return BuildInstanceHookEnvByObject(self, self.instance)
680

    
681
  def BuildHooksNodes(self):
682
    """Build hooks nodes.
683

684
    """
685
    nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
686
    return (nl, nl)
687

    
688
  def CheckPrereq(self):
689
    """Check prerequisites.
690

691
    This checks that the instance is in the cluster and is not running.
692

693
    """
694
    instance = self.cfg.GetInstanceInfo(self.op.instance_name)
695
    assert instance is not None, \
696
      "Cannot retrieve locked instance %s" % self.op.instance_name
697
    if self.op.nodes:
698
      if len(self.op.nodes) != len(instance.all_nodes):
699
        raise errors.OpPrereqError("Instance %s currently has %d nodes, but"
700
                                   " %d replacement nodes were specified" %
701
                                   (instance.name, len(instance.all_nodes),
702
                                    len(self.op.nodes)),
703
                                   errors.ECODE_INVAL)
704
      assert instance.disk_template != constants.DT_DRBD8 or \
705
             len(self.op.nodes) == 2
706
      assert instance.disk_template != constants.DT_PLAIN or \
707
             len(self.op.nodes) == 1
708
      primary_node = self.op.nodes[0]
709
    else:
710
      primary_node = instance.primary_node
711
    if not self.op.iallocator:
712
      CheckNodeOnline(self, primary_node)
713

    
714
    if instance.disk_template == constants.DT_DISKLESS:
715
      raise errors.OpPrereqError("Instance '%s' has no disks" %
716
                                 self.op.instance_name, errors.ECODE_INVAL)
717

    
718
    # Verify if node group locks are still correct
719
    owned_groups = self.owned_locks(locking.LEVEL_NODEGROUP)
720
    if owned_groups:
721
      # Node group locks are acquired only for the primary node (and only
722
      # when the allocator is used)
723
      CheckInstanceNodeGroups(self.cfg, self.op.instance_name, owned_groups,
724
                              primary_only=True)
725

    
726
    # if we replace nodes *and* the old primary is offline, we don't
727
    # check the instance state
728
    old_pnode = self.cfg.GetNodeInfo(instance.primary_node)
729
    if not ((self.op.iallocator or self.op.nodes) and old_pnode.offline):
730
      CheckInstanceState(self, instance, INSTANCE_NOT_RUNNING,
731
                         msg="cannot recreate disks")
732

    
733
    if self.op.disks:
734
      self.disks = dict(self.op.disks)
735
    else:
736
      self.disks = dict((idx, {}) for idx in range(len(instance.disks)))
737

    
738
    maxidx = max(self.disks.keys())
739
    if maxidx >= len(instance.disks):
740
      raise errors.OpPrereqError("Invalid disk index '%s'" % maxidx,
741
                                 errors.ECODE_INVAL)
742

    
743
    if ((self.op.nodes or self.op.iallocator) and
744
         sorted(self.disks.keys()) != range(len(instance.disks))):
745
      raise errors.OpPrereqError("Can't recreate disks partially and"
746
                                 " change the nodes at the same time",
747
                                 errors.ECODE_INVAL)
748

    
749
    self.instance = instance
750

    
751
    if self.op.iallocator:
752
      self._RunAllocator()
753
      # Release unneeded node and node resource locks
754
      ReleaseLocks(self, locking.LEVEL_NODE, keep=self.op.nodes)
755
      ReleaseLocks(self, locking.LEVEL_NODE_RES, keep=self.op.nodes)
756
      ReleaseLocks(self, locking.LEVEL_NODE_ALLOC)
757

    
758
    assert not self.glm.is_owned(locking.LEVEL_NODE_ALLOC)
759

    
760
  def Exec(self, feedback_fn):
761
    """Recreate the disks.
762

763
    """
764
    instance = self.instance
765

    
766
    assert (self.owned_locks(locking.LEVEL_NODE) ==
767
            self.owned_locks(locking.LEVEL_NODE_RES))
768

    
769
    to_skip = []
770
    mods = [] # keeps track of needed changes
771

    
772
    for idx, disk in enumerate(instance.disks):
773
      try:
774
        changes = self.disks[idx]
775
      except KeyError:
776
        # Disk should not be recreated
777
        to_skip.append(idx)
778
        continue
779

    
780
      # update secondaries for disks, if needed
781
      if self.op.nodes and disk.dev_type == constants.LD_DRBD8:
782
        # need to update the nodes and minors
783
        assert len(self.op.nodes) == 2
784
        assert len(disk.logical_id) == 6 # otherwise disk internals
785
                                         # have changed
786
        (_, _, old_port, _, _, old_secret) = disk.logical_id
787
        new_minors = self.cfg.AllocateDRBDMinor(self.op.nodes, instance.name)
788
        new_id = (self.op.nodes[0], self.op.nodes[1], old_port,
789
                  new_minors[0], new_minors[1], old_secret)
790
        assert len(disk.logical_id) == len(new_id)
791
      else:
792
        new_id = None
793

    
794
      mods.append((idx, new_id, changes))
795

    
796
    # now that we have passed all asserts above, we can apply the mods
797
    # in a single run (to avoid partial changes)
798
    for idx, new_id, changes in mods:
799
      disk = instance.disks[idx]
800
      if new_id is not None:
801
        assert disk.dev_type == constants.LD_DRBD8
802
        disk.logical_id = new_id
803
      if changes:
804
        disk.Update(size=changes.get(constants.IDISK_SIZE, None),
805
                    mode=changes.get(constants.IDISK_MODE, None))
806

    
807
    # change primary node, if needed
808
    if self.op.nodes:
809
      instance.primary_node = self.op.nodes[0]
810
      self.LogWarning("Changing the instance's nodes, you will have to"
811
                      " remove any disks left on the older nodes manually")
812

    
813
    if self.op.nodes:
814
      self.cfg.Update(instance, feedback_fn)
815

    
816
    # All touched nodes must be locked
817
    mylocks = self.owned_locks(locking.LEVEL_NODE)
818
    assert mylocks.issuperset(frozenset(instance.all_nodes))
819
    new_disks = CreateDisks(self, instance, to_skip=to_skip)
820

    
821
    # TODO: Release node locks before wiping, or explain why it's not possible
822
    if self.cfg.GetClusterInfo().prealloc_wipe_disks:
823
      wipedisks = [(idx, disk, 0)
824
                   for (idx, disk) in enumerate(instance.disks)
825
                   if idx not in to_skip]
826
      WipeOrCleanupDisks(self, instance, disks=wipedisks, cleanup=new_disks)
827

    
828

    
829
def _CheckNodesFreeDiskOnVG(lu, nodenames, vg, requested):
830
  """Checks if nodes have enough free disk space in the specified VG.
831

832
  This function checks if all given nodes have the needed amount of
833
  free disk. In case any node has less disk or we cannot get the
834
  information from the node, this function raises an OpPrereqError
835
  exception.
836

837
  @type lu: C{LogicalUnit}
838
  @param lu: a logical unit from which we get configuration data
839
  @type nodenames: C{list}
840
  @param nodenames: the list of node names to check
841
  @type vg: C{str}
842
  @param vg: the volume group to check
843
  @type requested: C{int}
844
  @param requested: the amount of disk in MiB to check for
845
  @raise errors.OpPrereqError: if the node doesn't have enough disk,
846
      or we cannot check the node
847

848
  """
849
  es_flags = rpc.GetExclusiveStorageForNodeNames(lu.cfg, nodenames)
850
  nodeinfo = lu.rpc.call_node_info(nodenames, [vg], None, es_flags)
851
  for node in nodenames:
852
    info = nodeinfo[node]
853
    info.Raise("Cannot get current information from node %s" % node,
854
               prereq=True, ecode=errors.ECODE_ENVIRON)
855
    (_, (vg_info, ), _) = info.payload
856
    vg_free = vg_info.get("vg_free", None)
857
    if not isinstance(vg_free, int):
858
      raise errors.OpPrereqError("Can't compute free disk space on node"
859
                                 " %s for vg %s, result was '%s'" %
860
                                 (node, vg, vg_free), errors.ECODE_ENVIRON)
861
    if requested > vg_free:
862
      raise errors.OpPrereqError("Not enough disk space on target node %s"
863
                                 " vg %s: required %d MiB, available %d MiB" %
864
                                 (node, vg, requested, vg_free),
865
                                 errors.ECODE_NORES)
866

    
867

    
868
def CheckNodesFreeDiskPerVG(lu, nodenames, req_sizes):
869
  """Checks if nodes have enough free disk space in all the VGs.
870

871
  This function checks if all given nodes have the needed amount of
872
  free disk. In case any node has less disk or we cannot get the
873
  information from the node, this function raises an OpPrereqError
874
  exception.
875

876
  @type lu: C{LogicalUnit}
877
  @param lu: a logical unit from which we get configuration data
878
  @type nodenames: C{list}
879
  @param nodenames: the list of node names to check
880
  @type req_sizes: C{dict}
881
  @param req_sizes: the hash of vg and corresponding amount of disk in
882
      MiB to check for
883
  @raise errors.OpPrereqError: if the node doesn't have enough disk,
884
      or we cannot check the node
885

886
  """
887
  for vg, req_size in req_sizes.items():
888
    _CheckNodesFreeDiskOnVG(lu, nodenames, vg, req_size)
889

    
890

    
891
def _DiskSizeInBytesToMebibytes(lu, size):
892
  """Converts a disk size in bytes to mebibytes.
893

894
  Warns and rounds up if the size isn't an even multiple of 1 MiB.
895

896
  """
897
  (mib, remainder) = divmod(size, 1024 * 1024)
898

    
899
  if remainder != 0:
900
    lu.LogWarning("Disk size is not an even multiple of 1 MiB; rounding up"
901
                  " to not overwrite existing data (%s bytes will not be"
902
                  " wiped)", (1024 * 1024) - remainder)
903
    mib += 1
904

    
905
  return mib
906

    
907

    
908
def _CalcEta(time_taken, written, total_size):
909
  """Calculates the ETA based on size written and total size.
910

911
  @param time_taken: The time taken so far
912
  @param written: amount written so far
913
  @param total_size: The total size of data to be written
914
  @return: The remaining time in seconds
915

916
  """
917
  avg_time = time_taken / float(written)
918
  return (total_size - written) * avg_time
919

    
920

    
921
def WipeDisks(lu, instance, disks=None):
922
  """Wipes instance disks.
923

924
  @type lu: L{LogicalUnit}
925
  @param lu: the logical unit on whose behalf we execute
926
  @type instance: L{objects.Instance}
927
  @param instance: the instance whose disks we should create
928
  @type disks: None or list of tuple of (number, L{objects.Disk}, number)
929
  @param disks: Disk details; tuple contains disk index, disk object and the
930
    start offset
931

932
  """
933
  node = instance.primary_node
934

    
935
  if disks is None:
936
    disks = [(idx, disk, 0)
937
             for (idx, disk) in enumerate(instance.disks)]
938

    
939
  for (_, device, _) in disks:
940
    lu.cfg.SetDiskID(device, node)
941

    
942
  logging.info("Pausing synchronization of disks of instance '%s'",
943
               instance.name)
944
  result = lu.rpc.call_blockdev_pause_resume_sync(node,
945
                                                  (map(compat.snd, disks),
946
                                                   instance),
947
                                                  True)
948
  result.Raise("Failed to pause disk synchronization on node '%s'" % node)
949

    
950
  for idx, success in enumerate(result.payload):
951
    if not success:
952
      logging.warn("Pausing synchronization of disk %s of instance '%s'"
953
                   " failed", idx, instance.name)
954

    
955
  try:
956
    for (idx, device, offset) in disks:
957
      # The wipe size is MIN_WIPE_CHUNK_PERCENT % of the instance disk but
958
      # MAX_WIPE_CHUNK at max. Truncating to integer to avoid rounding errors.
959
      wipe_chunk_size = \
960
        int(min(constants.MAX_WIPE_CHUNK,
961
                device.size / 100.0 * constants.MIN_WIPE_CHUNK_PERCENT))
962

    
963
      size = device.size
964
      last_output = 0
965
      start_time = time.time()
966

    
967
      if offset == 0:
968
        info_text = ""
969
      else:
970
        info_text = (" (from %s to %s)" %
971
                     (utils.FormatUnit(offset, "h"),
972
                      utils.FormatUnit(size, "h")))
973

    
974
      lu.LogInfo("* Wiping disk %s%s", idx, info_text)
975

    
976
      logging.info("Wiping disk %d for instance %s on node %s using"
977
                   " chunk size %s", idx, instance.name, node, wipe_chunk_size)
978

    
979
      while offset < size:
980
        wipe_size = min(wipe_chunk_size, size - offset)
981

    
982
        logging.debug("Wiping disk %d, offset %s, chunk %s",
983
                      idx, offset, wipe_size)
984

    
985
        result = lu.rpc.call_blockdev_wipe(node, (device, instance), offset,
986
                                           wipe_size)
987
        result.Raise("Could not wipe disk %d at offset %d for size %d" %
988
                     (idx, offset, wipe_size))
989

    
990
        now = time.time()
991
        offset += wipe_size
992
        if now - last_output >= 60:
993
          eta = _CalcEta(now - start_time, offset, size)
994
          lu.LogInfo(" - done: %.1f%% ETA: %s",
995
                     offset / float(size) * 100, utils.FormatSeconds(eta))
996
          last_output = now
997
  finally:
998
    logging.info("Resuming synchronization of disks for instance '%s'",
999
                 instance.name)
1000

    
1001
    result = lu.rpc.call_blockdev_pause_resume_sync(node,
1002
                                                    (map(compat.snd, disks),
1003
                                                     instance),
1004
                                                    False)
1005

    
1006
    if result.fail_msg:
1007
      lu.LogWarning("Failed to resume disk synchronization on node '%s': %s",
1008
                    node, result.fail_msg)
1009
    else:
1010
      for idx, success in enumerate(result.payload):
1011
        if not success:
1012
          lu.LogWarning("Resuming synchronization of disk %s of instance '%s'"
1013
                        " failed", idx, instance.name)
1014

    
1015

    
1016
def WipeOrCleanupDisks(lu, instance, disks=None, cleanup=None):
1017
  """Wrapper for L{WipeDisks} that handles errors.
1018

1019
  @type lu: L{LogicalUnit}
1020
  @param lu: the logical unit on whose behalf we execute
1021
  @type instance: L{objects.Instance}
1022
  @param instance: the instance whose disks we should wipe
1023
  @param disks: see L{WipeDisks}
1024
  @param cleanup: the result returned by L{CreateDisks}, used for cleanup in
1025
      case of error
1026
  @raise errors.OpPrereqError: in case of failure
1027

1028
  """
1029
  try:
1030
    WipeDisks(lu, instance, disks=disks)
1031
  except errors.OpExecError:
1032
    logging.warning("Wiping disks for instance '%s' failed",
1033
                    instance.name)
1034
    _UndoCreateDisks(lu, cleanup)
1035
    raise
1036

    
1037

    
1038
def ExpandCheckDisks(instance, disks):
1039
  """Return the instance disks selected by the disks list
1040

1041
  @type disks: list of L{objects.Disk} or None
1042
  @param disks: selected disks
1043
  @rtype: list of L{objects.Disk}
1044
  @return: selected instance disks to act on
1045

1046
  """
1047
  if disks is None:
1048
    return instance.disks
1049
  else:
1050
    if not set(disks).issubset(instance.disks):
1051
      raise errors.ProgrammerError("Can only act on disks belonging to the"
1052
                                   " target instance: expected a subset of %r,"
1053
                                   " got %r" % (instance.disks, disks))
1054
    return disks
1055

    
1056

    
1057
def WaitForSync(lu, instance, disks=None, oneshot=False):
1058
  """Sleep and poll for an instance's disk to sync.
1059

1060
  """
1061
  if not instance.disks or disks is not None and not disks:
1062
    return True
1063

    
1064
  disks = ExpandCheckDisks(instance, disks)
1065

    
1066
  if not oneshot:
1067
    lu.LogInfo("Waiting for instance %s to sync disks", instance.name)
1068

    
1069
  node = instance.primary_node
1070

    
1071
  for dev in disks:
1072
    lu.cfg.SetDiskID(dev, node)
1073

    
1074
  # TODO: Convert to utils.Retry
1075

    
1076
  retries = 0
1077
  degr_retries = 10 # in seconds, as we sleep 1 second each time
1078
  while True:
1079
    max_time = 0
1080
    done = True
1081
    cumul_degraded = False
1082
    rstats = lu.rpc.call_blockdev_getmirrorstatus(node, (disks, instance))
1083
    msg = rstats.fail_msg
1084
    if msg:
1085
      lu.LogWarning("Can't get any data from node %s: %s", node, msg)
1086
      retries += 1
1087
      if retries >= 10:
1088
        raise errors.RemoteError("Can't contact node %s for mirror data,"
1089
                                 " aborting." % node)
1090
      time.sleep(6)
1091
      continue
1092
    rstats = rstats.payload
1093
    retries = 0
1094
    for i, mstat in enumerate(rstats):
1095
      if mstat is None:
1096
        lu.LogWarning("Can't compute data for node %s/%s",
1097
                      node, disks[i].iv_name)
1098
        continue
1099

    
1100
      cumul_degraded = (cumul_degraded or
1101
                        (mstat.is_degraded and mstat.sync_percent is None))
1102
      if mstat.sync_percent is not None:
1103
        done = False
1104
        if mstat.estimated_time is not None:
1105
          rem_time = ("%s remaining (estimated)" %
1106
                      utils.FormatSeconds(mstat.estimated_time))
1107
          max_time = mstat.estimated_time
1108
        else:
1109
          rem_time = "no time estimate"
1110
        lu.LogInfo("- device %s: %5.2f%% done, %s",
1111
                   disks[i].iv_name, mstat.sync_percent, rem_time)
1112

    
1113
    # if we're done but degraded, let's do a few small retries, to
1114
    # make sure we see a stable and not transient situation; therefore
1115
    # we force restart of the loop
1116
    if (done or oneshot) and cumul_degraded and degr_retries > 0:
1117
      logging.info("Degraded disks found, %d retries left", degr_retries)
1118
      degr_retries -= 1
1119
      time.sleep(1)
1120
      continue
1121

    
1122
    if done or oneshot:
1123
      break
1124

    
1125
    time.sleep(min(60, max_time))
1126

    
1127
  if done:
1128
    lu.LogInfo("Instance %s's disks are in sync", instance.name)
1129

    
1130
  return not cumul_degraded
1131

    
1132

    
1133
def ShutdownInstanceDisks(lu, instance, disks=None, ignore_primary=False):
1134
  """Shutdown block devices of an instance.
1135

1136
  This does the shutdown on all nodes of the instance.
1137

1138
  If the ignore_primary is false, errors on the primary node are
1139
  ignored.
1140

1141
  """
1142
  lu.cfg.MarkInstanceDisksInactive(instance.name)
1143
  all_result = True
1144
  disks = ExpandCheckDisks(instance, disks)
1145

    
1146
  for disk in disks:
1147
    for node, top_disk in disk.ComputeNodeTree(instance.primary_node):
1148
      lu.cfg.SetDiskID(top_disk, node)
1149
      result = lu.rpc.call_blockdev_shutdown(node, (top_disk, instance))
1150
      msg = result.fail_msg
1151
      if msg:
1152
        lu.LogWarning("Could not shutdown block device %s on node %s: %s",
1153
                      disk.iv_name, node, msg)
1154
        if ((node == instance.primary_node and not ignore_primary) or
1155
            (node != instance.primary_node and not result.offline)):
1156
          all_result = False
1157
  return all_result
1158

    
1159

    
1160
def _SafeShutdownInstanceDisks(lu, instance, disks=None):
1161
  """Shutdown block devices of an instance.
1162

1163
  This function checks if an instance is running, before calling
1164
  _ShutdownInstanceDisks.
1165

1166
  """
1167
  CheckInstanceState(lu, instance, INSTANCE_DOWN, msg="cannot shutdown disks")
1168
  ShutdownInstanceDisks(lu, instance, disks=disks)
1169

    
1170

    
1171
def AssembleInstanceDisks(lu, instance, disks=None, ignore_secondaries=False,
1172
                           ignore_size=False):
1173
  """Prepare the block devices for an instance.
1174

1175
  This sets up the block devices on all nodes.
1176

1177
  @type lu: L{LogicalUnit}
1178
  @param lu: the logical unit on whose behalf we execute
1179
  @type instance: L{objects.Instance}
1180
  @param instance: the instance for whose disks we assemble
1181
  @type disks: list of L{objects.Disk} or None
1182
  @param disks: which disks to assemble (or all, if None)
1183
  @type ignore_secondaries: boolean
1184
  @param ignore_secondaries: if true, errors on secondary nodes
1185
      won't result in an error return from the function
1186
  @type ignore_size: boolean
1187
  @param ignore_size: if true, the current known size of the disk
1188
      will not be used during the disk activation, useful for cases
1189
      when the size is wrong
1190
  @return: False if the operation failed, otherwise a list of
1191
      (host, instance_visible_name, node_visible_name)
1192
      with the mapping from node devices to instance devices
1193

1194
  """
1195
  device_info = []
1196
  disks_ok = True
1197
  iname = instance.name
1198
  disks = ExpandCheckDisks(instance, disks)
1199

    
1200
  # With the two passes mechanism we try to reduce the window of
1201
  # opportunity for the race condition of switching DRBD to primary
1202
  # before handshaking occured, but we do not eliminate it
1203

    
1204
  # The proper fix would be to wait (with some limits) until the
1205
  # connection has been made and drbd transitions from WFConnection
1206
  # into any other network-connected state (Connected, SyncTarget,
1207
  # SyncSource, etc.)
1208

    
1209
  # mark instance disks as active before doing actual work, so watcher does
1210
  # not try to shut them down erroneously
1211
  lu.cfg.MarkInstanceDisksActive(iname)
1212

    
1213
  # 1st pass, assemble on all nodes in secondary mode
1214
  for idx, inst_disk in enumerate(disks):
1215
    for node, node_disk in inst_disk.ComputeNodeTree(instance.primary_node):
1216
      if ignore_size:
1217
        node_disk = node_disk.Copy()
1218
        node_disk.UnsetSize()
1219
      lu.cfg.SetDiskID(node_disk, node)
1220
      result = lu.rpc.call_blockdev_assemble(node, (node_disk, instance), iname,
1221
                                             False, idx)
1222
      msg = result.fail_msg
1223
      if msg:
1224
        is_offline_secondary = (node in instance.secondary_nodes and
1225
                                result.offline)
1226
        lu.LogWarning("Could not prepare block device %s on node %s"
1227
                      " (is_primary=False, pass=1): %s",
1228
                      inst_disk.iv_name, node, msg)
1229
        if not (ignore_secondaries or is_offline_secondary):
1230
          disks_ok = False
1231

    
1232
  # FIXME: race condition on drbd migration to primary
1233

    
1234
  # 2nd pass, do only the primary node
1235
  for idx, inst_disk in enumerate(disks):
1236
    dev_path = None
1237

    
1238
    for node, node_disk in inst_disk.ComputeNodeTree(instance.primary_node):
1239
      if node != instance.primary_node:
1240
        continue
1241
      if ignore_size:
1242
        node_disk = node_disk.Copy()
1243
        node_disk.UnsetSize()
1244
      lu.cfg.SetDiskID(node_disk, node)
1245
      result = lu.rpc.call_blockdev_assemble(node, (node_disk, instance), iname,
1246
                                             True, idx)
1247
      msg = result.fail_msg
1248
      if msg:
1249
        lu.LogWarning("Could not prepare block device %s on node %s"
1250
                      " (is_primary=True, pass=2): %s",
1251
                      inst_disk.iv_name, node, msg)
1252
        disks_ok = False
1253
      else:
1254
        dev_path = result.payload
1255

    
1256
    device_info.append((instance.primary_node, inst_disk.iv_name, dev_path))
1257

    
1258
  # leave the disks configured for the primary node
1259
  # this is a workaround that would be fixed better by
1260
  # improving the logical/physical id handling
1261
  for disk in disks:
1262
    lu.cfg.SetDiskID(disk, instance.primary_node)
1263

    
1264
  if not disks_ok:
1265
    lu.cfg.MarkInstanceDisksInactive(iname)
1266

    
1267
  return disks_ok, device_info
1268

    
1269

    
1270
def StartInstanceDisks(lu, instance, force):
1271
  """Start the disks of an instance.
1272

1273
  """
1274
  disks_ok, _ = AssembleInstanceDisks(lu, instance,
1275
                                      ignore_secondaries=force)
1276
  if not disks_ok:
1277
    ShutdownInstanceDisks(lu, instance)
1278
    if force is not None and not force:
1279
      lu.LogWarning("",
1280
                    hint=("If the message above refers to a secondary node,"
1281
                          " you can retry the operation using '--force'"))
1282
    raise errors.OpExecError("Disk consistency error")
1283

    
1284

    
1285
class LUInstanceGrowDisk(LogicalUnit):
1286
  """Grow a disk of an instance.
1287

1288
  """
1289
  HPATH = "disk-grow"
1290
  HTYPE = constants.HTYPE_INSTANCE
1291
  REQ_BGL = False
1292

    
1293
  def ExpandNames(self):
1294
    self._ExpandAndLockInstance()
1295
    self.needed_locks[locking.LEVEL_NODE] = []
1296
    self.needed_locks[locking.LEVEL_NODE_RES] = []
1297
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
1298
    self.recalculate_locks[locking.LEVEL_NODE_RES] = constants.LOCKS_REPLACE
1299

    
1300
  def DeclareLocks(self, level):
1301
    if level == locking.LEVEL_NODE:
1302
      self._LockInstancesNodes()
1303
    elif level == locking.LEVEL_NODE_RES:
1304
      # Copy node locks
1305
      self.needed_locks[locking.LEVEL_NODE_RES] = \
1306
        CopyLockList(self.needed_locks[locking.LEVEL_NODE])
1307

    
1308
  def BuildHooksEnv(self):
1309
    """Build hooks env.
1310

1311
    This runs on the master, the primary and all the secondaries.
1312

1313
    """
1314
    env = {
1315
      "DISK": self.op.disk,
1316
      "AMOUNT": self.op.amount,
1317
      "ABSOLUTE": self.op.absolute,
1318
      }
1319
    env.update(BuildInstanceHookEnvByObject(self, self.instance))
1320
    return env
1321

    
1322
  def BuildHooksNodes(self):
1323
    """Build hooks nodes.
1324

1325
    """
1326
    nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
1327
    return (nl, nl)
1328

    
1329
  def CheckPrereq(self):
1330
    """Check prerequisites.
1331

1332
    This checks that the instance is in the cluster.
1333

1334
    """
1335
    instance = self.cfg.GetInstanceInfo(self.op.instance_name)
1336
    assert instance is not None, \
1337
      "Cannot retrieve locked instance %s" % self.op.instance_name
1338
    nodenames = list(instance.all_nodes)
1339
    for node in nodenames:
1340
      CheckNodeOnline(self, node)
1341

    
1342
    self.instance = instance
1343

    
1344
    if instance.disk_template not in constants.DTS_GROWABLE:
1345
      raise errors.OpPrereqError("Instance's disk layout does not support"
1346
                                 " growing", errors.ECODE_INVAL)
1347

    
1348
    self.disk = instance.FindDisk(self.op.disk)
1349

    
1350
    if self.op.absolute:
1351
      self.target = self.op.amount
1352
      self.delta = self.target - self.disk.size
1353
      if self.delta < 0:
1354
        raise errors.OpPrereqError("Requested size (%s) is smaller than "
1355
                                   "current disk size (%s)" %
1356
                                   (utils.FormatUnit(self.target, "h"),
1357
                                    utils.FormatUnit(self.disk.size, "h")),
1358
                                   errors.ECODE_STATE)
1359
    else:
1360
      self.delta = self.op.amount
1361
      self.target = self.disk.size + self.delta
1362
      if self.delta < 0:
1363
        raise errors.OpPrereqError("Requested increment (%s) is negative" %
1364
                                   utils.FormatUnit(self.delta, "h"),
1365
                                   errors.ECODE_INVAL)
1366

    
1367
    self._CheckDiskSpace(nodenames, self.disk.ComputeGrowth(self.delta))
1368

    
1369
  def _CheckDiskSpace(self, nodenames, req_vgspace):
1370
    template = self.instance.disk_template
1371
    if template not in (constants.DTS_NO_FREE_SPACE_CHECK):
1372
      # TODO: check the free disk space for file, when that feature will be
1373
      # supported
1374
      nodes = map(self.cfg.GetNodeInfo, nodenames)
1375
      es_nodes = filter(lambda n: IsExclusiveStorageEnabledNode(self.cfg, n),
1376
                        nodes)
1377
      if es_nodes:
1378
        # With exclusive storage we need to something smarter than just looking
1379
        # at free space; for now, let's simply abort the operation.
1380
        raise errors.OpPrereqError("Cannot grow disks when exclusive_storage"
1381
                                   " is enabled", errors.ECODE_STATE)
1382
      CheckNodesFreeDiskPerVG(self, nodenames, req_vgspace)
1383

    
1384
  def Exec(self, feedback_fn):
1385
    """Execute disk grow.
1386

1387
    """
1388
    instance = self.instance
1389
    disk = self.disk
1390

    
1391
    assert set([instance.name]) == self.owned_locks(locking.LEVEL_INSTANCE)
1392
    assert (self.owned_locks(locking.LEVEL_NODE) ==
1393
            self.owned_locks(locking.LEVEL_NODE_RES))
1394

    
1395
    wipe_disks = self.cfg.GetClusterInfo().prealloc_wipe_disks
1396

    
1397
    disks_ok, _ = AssembleInstanceDisks(self, self.instance, disks=[disk])
1398
    if not disks_ok:
1399
      raise errors.OpExecError("Cannot activate block device to grow")
1400

    
1401
    feedback_fn("Growing disk %s of instance '%s' by %s to %s" %
1402
                (self.op.disk, instance.name,
1403
                 utils.FormatUnit(self.delta, "h"),
1404
                 utils.FormatUnit(self.target, "h")))
1405

    
1406
    # First run all grow ops in dry-run mode
1407
    for node in instance.all_nodes:
1408
      self.cfg.SetDiskID(disk, node)
1409
      result = self.rpc.call_blockdev_grow(node, (disk, instance), self.delta,
1410
                                           True, True)
1411
      result.Raise("Dry-run grow request failed to node %s" % node)
1412

    
1413
    if wipe_disks:
1414
      # Get disk size from primary node for wiping
1415
      self.cfg.SetDiskID(disk, instance.primary_node)
1416
      result = self.rpc.call_blockdev_getsize(instance.primary_node, [disk])
1417
      result.Raise("Failed to retrieve disk size from node '%s'" %
1418
                   instance.primary_node)
1419

    
1420
      (disk_size_in_bytes, ) = result.payload
1421

    
1422
      if disk_size_in_bytes is None:
1423
        raise errors.OpExecError("Failed to retrieve disk size from primary"
1424
                                 " node '%s'" % instance.primary_node)
1425

    
1426
      old_disk_size = _DiskSizeInBytesToMebibytes(self, disk_size_in_bytes)
1427

    
1428
      assert old_disk_size >= disk.size, \
1429
        ("Retrieved disk size too small (got %s, should be at least %s)" %
1430
         (old_disk_size, disk.size))
1431
    else:
1432
      old_disk_size = None
1433

    
1434
    # We know that (as far as we can test) operations across different
1435
    # nodes will succeed, time to run it for real on the backing storage
1436
    for node in instance.all_nodes:
1437
      self.cfg.SetDiskID(disk, node)
1438
      result = self.rpc.call_blockdev_grow(node, (disk, instance), self.delta,
1439
                                           False, True)
1440
      result.Raise("Grow request failed to node %s" % node)
1441

    
1442
    # And now execute it for logical storage, on the primary node
1443
    node = instance.primary_node
1444
    self.cfg.SetDiskID(disk, node)
1445
    result = self.rpc.call_blockdev_grow(node, (disk, instance), self.delta,
1446
                                         False, False)
1447
    result.Raise("Grow request failed to node %s" % node)
1448

    
1449
    disk.RecordGrow(self.delta)
1450
    self.cfg.Update(instance, feedback_fn)
1451

    
1452
    # Changes have been recorded, release node lock
1453
    ReleaseLocks(self, locking.LEVEL_NODE)
1454

    
1455
    # Downgrade lock while waiting for sync
1456
    self.glm.downgrade(locking.LEVEL_INSTANCE)
1457

    
1458
    assert wipe_disks ^ (old_disk_size is None)
1459

    
1460
    if wipe_disks:
1461
      assert instance.disks[self.op.disk] == disk
1462

    
1463
      # Wipe newly added disk space
1464
      WipeDisks(self, instance,
1465
                disks=[(self.op.disk, disk, old_disk_size)])
1466

    
1467
    if self.op.wait_for_sync:
1468
      disk_abort = not WaitForSync(self, instance, disks=[disk])
1469
      if disk_abort:
1470
        self.LogWarning("Disk syncing has not returned a good status; check"
1471
                        " the instance")
1472
      if not instance.disks_active:
1473
        _SafeShutdownInstanceDisks(self, instance, disks=[disk])
1474
    elif not instance.disks_active:
1475
      self.LogWarning("Not shutting down the disk even if the instance is"
1476
                      " not supposed to be running because no wait for"
1477
                      " sync mode was requested")
1478

    
1479
    assert self.owned_locks(locking.LEVEL_NODE_RES)
1480
    assert set([instance.name]) == self.owned_locks(locking.LEVEL_INSTANCE)
1481

    
1482

    
1483
class LUInstanceReplaceDisks(LogicalUnit):
1484
  """Replace the disks of an instance.
1485

1486
  """
1487
  HPATH = "mirrors-replace"
1488
  HTYPE = constants.HTYPE_INSTANCE
1489
  REQ_BGL = False
1490

    
1491
  def CheckArguments(self):
1492
    """Check arguments.
1493

1494
    """
1495
    remote_node = self.op.remote_node
1496
    ialloc = self.op.iallocator
1497
    if self.op.mode == constants.REPLACE_DISK_CHG:
1498
      if remote_node is None and ialloc is None:
1499
        raise errors.OpPrereqError("When changing the secondary either an"
1500
                                   " iallocator script must be used or the"
1501
                                   " new node given", errors.ECODE_INVAL)
1502
      else:
1503
        CheckIAllocatorOrNode(self, "iallocator", "remote_node")
1504

    
1505
    elif remote_node is not None or ialloc is not None:
1506
      # Not replacing the secondary
1507
      raise errors.OpPrereqError("The iallocator and new node options can"
1508
                                 " only be used when changing the"
1509
                                 " secondary node", errors.ECODE_INVAL)
1510

    
1511
  def ExpandNames(self):
1512
    self._ExpandAndLockInstance()
1513

    
1514
    assert locking.LEVEL_NODE not in self.needed_locks
1515
    assert locking.LEVEL_NODE_RES not in self.needed_locks
1516
    assert locking.LEVEL_NODEGROUP not in self.needed_locks
1517

    
1518
    assert self.op.iallocator is None or self.op.remote_node is None, \
1519
      "Conflicting options"
1520

    
1521
    if self.op.remote_node is not None:
1522
      self.op.remote_node = ExpandNodeName(self.cfg, self.op.remote_node)
1523

    
1524
      # Warning: do not remove the locking of the new secondary here
1525
      # unless DRBD8.AddChildren is changed to work in parallel;
1526
      # currently it doesn't since parallel invocations of
1527
      # FindUnusedMinor will conflict
1528
      self.needed_locks[locking.LEVEL_NODE] = [self.op.remote_node]
1529
      self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_APPEND
1530
    else:
1531
      self.needed_locks[locking.LEVEL_NODE] = []
1532
      self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
1533

    
1534
      if self.op.iallocator is not None:
1535
        # iallocator will select a new node in the same group
1536
        self.needed_locks[locking.LEVEL_NODEGROUP] = []
1537
        self.needed_locks[locking.LEVEL_NODE_ALLOC] = locking.ALL_SET
1538

    
1539
    self.needed_locks[locking.LEVEL_NODE_RES] = []
1540

    
1541
    self.replacer = TLReplaceDisks(self, self.op.instance_name, self.op.mode,
1542
                                   self.op.iallocator, self.op.remote_node,
1543
                                   self.op.disks, self.op.early_release,
1544
                                   self.op.ignore_ipolicy)
1545

    
1546
    self.tasklets = [self.replacer]
1547

    
1548
  def DeclareLocks(self, level):
1549
    if level == locking.LEVEL_NODEGROUP:
1550
      assert self.op.remote_node is None
1551
      assert self.op.iallocator is not None
1552
      assert not self.needed_locks[locking.LEVEL_NODEGROUP]
1553

    
1554
      self.share_locks[locking.LEVEL_NODEGROUP] = 1
1555
      # Lock all groups used by instance optimistically; this requires going
1556
      # via the node before it's locked, requiring verification later on
1557
      self.needed_locks[locking.LEVEL_NODEGROUP] = \
1558
        self.cfg.GetInstanceNodeGroups(self.op.instance_name)
1559

    
1560
    elif level == locking.LEVEL_NODE:
1561
      if self.op.iallocator is not None:
1562
        assert self.op.remote_node is None
1563
        assert not self.needed_locks[locking.LEVEL_NODE]
1564
        assert locking.NAL in self.owned_locks(locking.LEVEL_NODE_ALLOC)
1565

    
1566
        # Lock member nodes of all locked groups
1567
        self.needed_locks[locking.LEVEL_NODE] = \
1568
          [node_name
1569
           for group_uuid in self.owned_locks(locking.LEVEL_NODEGROUP)
1570
           for node_name in self.cfg.GetNodeGroup(group_uuid).members]
1571
      else:
1572
        assert not self.glm.is_owned(locking.LEVEL_NODE_ALLOC)
1573

    
1574
        self._LockInstancesNodes()
1575

    
1576
    elif level == locking.LEVEL_NODE_RES:
1577
      # Reuse node locks
1578
      self.needed_locks[locking.LEVEL_NODE_RES] = \
1579
        self.needed_locks[locking.LEVEL_NODE]
1580

    
1581
  def BuildHooksEnv(self):
1582
    """Build hooks env.
1583

1584
    This runs on the master, the primary and all the secondaries.
1585

1586
    """
1587
    instance = self.replacer.instance
1588
    env = {
1589
      "MODE": self.op.mode,
1590
      "NEW_SECONDARY": self.op.remote_node,
1591
      "OLD_SECONDARY": instance.secondary_nodes[0],
1592
      }
1593
    env.update(BuildInstanceHookEnvByObject(self, instance))
1594
    return env
1595

    
1596
  def BuildHooksNodes(self):
1597
    """Build hooks nodes.
1598

1599
    """
1600
    instance = self.replacer.instance
1601
    nl = [
1602
      self.cfg.GetMasterNode(),
1603
      instance.primary_node,
1604
      ]
1605
    if self.op.remote_node is not None:
1606
      nl.append(self.op.remote_node)
1607
    return nl, nl
1608

    
1609
  def CheckPrereq(self):
1610
    """Check prerequisites.
1611

1612
    """
1613
    assert (self.glm.is_owned(locking.LEVEL_NODEGROUP) or
1614
            self.op.iallocator is None)
1615

    
1616
    # Verify if node group locks are still correct
1617
    owned_groups = self.owned_locks(locking.LEVEL_NODEGROUP)
1618
    if owned_groups:
1619
      CheckInstanceNodeGroups(self.cfg, self.op.instance_name, owned_groups)
1620

    
1621
    return LogicalUnit.CheckPrereq(self)
1622

    
1623

    
1624
class LUInstanceActivateDisks(NoHooksLU):
1625
  """Bring up an instance's disks.
1626

1627
  """
1628
  REQ_BGL = False
1629

    
1630
  def ExpandNames(self):
1631
    self._ExpandAndLockInstance()
1632
    self.needed_locks[locking.LEVEL_NODE] = []
1633
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
1634

    
1635
  def DeclareLocks(self, level):
1636
    if level == locking.LEVEL_NODE:
1637
      self._LockInstancesNodes()
1638

    
1639
  def CheckPrereq(self):
1640
    """Check prerequisites.
1641

1642
    This checks that the instance is in the cluster.
1643

1644
    """
1645
    self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
1646
    assert self.instance is not None, \
1647
      "Cannot retrieve locked instance %s" % self.op.instance_name
1648
    CheckNodeOnline(self, self.instance.primary_node)
1649

    
1650
  def Exec(self, feedback_fn):
1651
    """Activate the disks.
1652

1653
    """
1654
    disks_ok, disks_info = \
1655
              AssembleInstanceDisks(self, self.instance,
1656
                                    ignore_size=self.op.ignore_size)
1657
    if not disks_ok:
1658
      raise errors.OpExecError("Cannot activate block devices")
1659

    
1660
    if self.op.wait_for_sync:
1661
      if not WaitForSync(self, self.instance):
1662
        self.cfg.MarkInstanceDisksInactive(self.instance.name)
1663
        raise errors.OpExecError("Some disks of the instance are degraded!")
1664

    
1665
    return disks_info
1666

    
1667

    
1668
class LUInstanceDeactivateDisks(NoHooksLU):
1669
  """Shutdown an instance's disks.
1670

1671
  """
1672
  REQ_BGL = False
1673

    
1674
  def ExpandNames(self):
1675
    self._ExpandAndLockInstance()
1676
    self.needed_locks[locking.LEVEL_NODE] = []
1677
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
1678

    
1679
  def DeclareLocks(self, level):
1680
    if level == locking.LEVEL_NODE:
1681
      self._LockInstancesNodes()
1682

    
1683
  def CheckPrereq(self):
1684
    """Check prerequisites.
1685

1686
    This checks that the instance is in the cluster.
1687

1688
    """
1689
    self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
1690
    assert self.instance is not None, \
1691
      "Cannot retrieve locked instance %s" % self.op.instance_name
1692

    
1693
  def Exec(self, feedback_fn):
1694
    """Deactivate the disks
1695

1696
    """
1697
    instance = self.instance
1698
    if self.op.force:
1699
      ShutdownInstanceDisks(self, instance)
1700
    else:
1701
      _SafeShutdownInstanceDisks(self, instance)
1702

    
1703

    
1704
def _CheckDiskConsistencyInner(lu, instance, dev, node, on_primary,
1705
                               ldisk=False):
1706
  """Check that mirrors are not degraded.
1707

1708
  @attention: The device has to be annotated already.
1709

1710
  The ldisk parameter, if True, will change the test from the
1711
  is_degraded attribute (which represents overall non-ok status for
1712
  the device(s)) to the ldisk (representing the local storage status).
1713

1714
  """
1715
  lu.cfg.SetDiskID(dev, node)
1716

    
1717
  result = True
1718

    
1719
  if on_primary or dev.AssembleOnSecondary():
1720
    rstats = lu.rpc.call_blockdev_find(node, dev)
1721
    msg = rstats.fail_msg
1722
    if msg:
1723
      lu.LogWarning("Can't find disk on node %s: %s", node, msg)
1724
      result = False
1725
    elif not rstats.payload:
1726
      lu.LogWarning("Can't find disk on node %s", node)
1727
      result = False
1728
    else:
1729
      if ldisk:
1730
        result = result and rstats.payload.ldisk_status == constants.LDS_OKAY
1731
      else:
1732
        result = result and not rstats.payload.is_degraded
1733

    
1734
  if dev.children:
1735
    for child in dev.children:
1736
      result = result and _CheckDiskConsistencyInner(lu, instance, child, node,
1737
                                                     on_primary)
1738

    
1739
  return result
1740

    
1741

    
1742
def CheckDiskConsistency(lu, instance, dev, node, on_primary, ldisk=False):
1743
  """Wrapper around L{_CheckDiskConsistencyInner}.
1744

1745
  """
1746
  (disk,) = AnnotateDiskParams(instance, [dev], lu.cfg)
1747
  return _CheckDiskConsistencyInner(lu, instance, disk, node, on_primary,
1748
                                    ldisk=ldisk)
1749

    
1750

    
1751
def _BlockdevFind(lu, node, dev, instance):
1752
  """Wrapper around call_blockdev_find to annotate diskparams.
1753

1754
  @param lu: A reference to the lu object
1755
  @param node: The node to call out
1756
  @param dev: The device to find
1757
  @param instance: The instance object the device belongs to
1758
  @returns The result of the rpc call
1759

1760
  """
1761
  (disk,) = AnnotateDiskParams(instance, [dev], lu.cfg)
1762
  return lu.rpc.call_blockdev_find(node, disk)
1763

    
1764

    
1765
def _GenerateUniqueNames(lu, exts):
1766
  """Generate a suitable LV name.
1767

1768
  This will generate a logical volume name for the given instance.
1769

1770
  """
1771
  results = []
1772
  for val in exts:
1773
    new_id = lu.cfg.GenerateUniqueID(lu.proc.GetECId())
1774
    results.append("%s%s" % (new_id, val))
1775
  return results
1776

    
1777

    
1778
class TLReplaceDisks(Tasklet):
1779
  """Replaces disks for an instance.
1780

1781
  Note: Locking is not within the scope of this class.
1782

1783
  """
1784
  def __init__(self, lu, instance_name, mode, iallocator_name, remote_node,
1785
               disks, early_release, ignore_ipolicy):
1786
    """Initializes this class.
1787

1788
    """
1789
    Tasklet.__init__(self, lu)
1790

    
1791
    # Parameters
1792
    self.instance_name = instance_name
1793
    self.mode = mode
1794
    self.iallocator_name = iallocator_name
1795
    self.remote_node = remote_node
1796
    self.disks = disks
1797
    self.early_release = early_release
1798
    self.ignore_ipolicy = ignore_ipolicy
1799

    
1800
    # Runtime data
1801
    self.instance = None
1802
    self.new_node = None
1803
    self.target_node = None
1804
    self.other_node = None
1805
    self.remote_node_info = None
1806
    self.node_secondary_ip = None
1807

    
1808
  @staticmethod
1809
  def _RunAllocator(lu, iallocator_name, instance_name, relocate_from):
1810
    """Compute a new secondary node using an IAllocator.
1811

1812
    """
1813
    req = iallocator.IAReqRelocate(name=instance_name,
1814
                                   relocate_from=list(relocate_from))
1815
    ial = iallocator.IAllocator(lu.cfg, lu.rpc, req)
1816

    
1817
    ial.Run(iallocator_name)
1818

    
1819
    if not ial.success:
1820
      raise errors.OpPrereqError("Can't compute nodes using iallocator '%s':"
1821
                                 " %s" % (iallocator_name, ial.info),
1822
                                 errors.ECODE_NORES)
1823

    
1824
    remote_node_name = ial.result[0]
1825

    
1826
    lu.LogInfo("Selected new secondary for instance '%s': %s",
1827
               instance_name, remote_node_name)
1828

    
1829
    return remote_node_name
1830

    
1831
  def _FindFaultyDisks(self, node_name):
1832
    """Wrapper for L{FindFaultyInstanceDisks}.
1833

1834
    """
1835
    return FindFaultyInstanceDisks(self.cfg, self.rpc, self.instance,
1836
                                   node_name, True)
1837

    
1838
  def _CheckDisksActivated(self, instance):
1839
    """Checks if the instance disks are activated.
1840

1841
    @param instance: The instance to check disks
1842
    @return: True if they are activated, False otherwise
1843

1844
    """
1845
    nodes = instance.all_nodes
1846

    
1847
    for idx, dev in enumerate(instance.disks):
1848
      for node in nodes:
1849
        self.lu.LogInfo("Checking disk/%d on %s", idx, node)
1850
        self.cfg.SetDiskID(dev, node)
1851

    
1852
        result = _BlockdevFind(self, node, dev, instance)
1853

    
1854
        if result.offline:
1855
          continue
1856
        elif result.fail_msg or not result.payload:
1857
          return False
1858

    
1859
    return True
1860

    
1861
  def CheckPrereq(self):
1862
    """Check prerequisites.
1863

1864
    This checks that the instance is in the cluster.
1865

1866
    """
1867
    self.instance = instance = self.cfg.GetInstanceInfo(self.instance_name)
1868
    assert instance is not None, \
1869
      "Cannot retrieve locked instance %s" % self.instance_name
1870

    
1871
    if instance.disk_template != constants.DT_DRBD8:
1872
      raise errors.OpPrereqError("Can only run replace disks for DRBD8-based"
1873
                                 " instances", errors.ECODE_INVAL)
1874

    
1875
    if len(instance.secondary_nodes) != 1:
1876
      raise errors.OpPrereqError("The instance has a strange layout,"
1877
                                 " expected one secondary but found %d" %
1878
                                 len(instance.secondary_nodes),
1879
                                 errors.ECODE_FAULT)
1880

    
1881
    instance = self.instance
1882
    secondary_node = instance.secondary_nodes[0]
1883

    
1884
    if self.iallocator_name is None:
1885
      remote_node = self.remote_node
1886
    else:
1887
      remote_node = self._RunAllocator(self.lu, self.iallocator_name,
1888
                                       instance.name, instance.secondary_nodes)
1889

    
1890
    if remote_node is None:
1891
      self.remote_node_info = None
1892
    else:
1893
      assert remote_node in self.lu.owned_locks(locking.LEVEL_NODE), \
1894
             "Remote node '%s' is not locked" % remote_node
1895

    
1896
      self.remote_node_info = self.cfg.GetNodeInfo(remote_node)
1897
      assert self.remote_node_info is not None, \
1898
        "Cannot retrieve locked node %s" % remote_node
1899

    
1900
    if remote_node == self.instance.primary_node:
1901
      raise errors.OpPrereqError("The specified node is the primary node of"
1902
                                 " the instance", errors.ECODE_INVAL)
1903

    
1904
    if remote_node == secondary_node:
1905
      raise errors.OpPrereqError("The specified node is already the"
1906
                                 " secondary node of the instance",
1907
                                 errors.ECODE_INVAL)
1908

    
1909
    if self.disks and self.mode in (constants.REPLACE_DISK_AUTO,
1910
                                    constants.REPLACE_DISK_CHG):
1911
      raise errors.OpPrereqError("Cannot specify disks to be replaced",
1912
                                 errors.ECODE_INVAL)
1913

    
1914
    if self.mode == constants.REPLACE_DISK_AUTO:
1915
      if not self._CheckDisksActivated(instance):
1916
        raise errors.OpPrereqError("Please run activate-disks on instance %s"
1917
                                   " first" % self.instance_name,
1918
                                   errors.ECODE_STATE)
1919
      faulty_primary = self._FindFaultyDisks(instance.primary_node)
1920
      faulty_secondary = self._FindFaultyDisks(secondary_node)
1921

    
1922
      if faulty_primary and faulty_secondary:
1923
        raise errors.OpPrereqError("Instance %s has faulty disks on more than"
1924
                                   " one node and can not be repaired"
1925
                                   " automatically" % self.instance_name,
1926
                                   errors.ECODE_STATE)
1927

    
1928
      if faulty_primary:
1929
        self.disks = faulty_primary
1930
        self.target_node = instance.primary_node
1931
        self.other_node = secondary_node
1932
        check_nodes = [self.target_node, self.other_node]
1933
      elif faulty_secondary:
1934
        self.disks = faulty_secondary
1935
        self.target_node = secondary_node
1936
        self.other_node = instance.primary_node
1937
        check_nodes = [self.target_node, self.other_node]
1938
      else:
1939
        self.disks = []
1940
        check_nodes = []
1941

    
1942
    else:
1943
      # Non-automatic modes
1944
      if self.mode == constants.REPLACE_DISK_PRI:
1945
        self.target_node = instance.primary_node
1946
        self.other_node = secondary_node
1947
        check_nodes = [self.target_node, self.other_node]
1948

    
1949
      elif self.mode == constants.REPLACE_DISK_SEC:
1950
        self.target_node = secondary_node
1951
        self.other_node = instance.primary_node
1952
        check_nodes = [self.target_node, self.other_node]
1953

    
1954
      elif self.mode == constants.REPLACE_DISK_CHG:
1955
        self.new_node = remote_node
1956
        self.other_node = instance.primary_node
1957
        self.target_node = secondary_node
1958
        check_nodes = [self.new_node, self.other_node]
1959

    
1960
        CheckNodeNotDrained(self.lu, remote_node)
1961
        CheckNodeVmCapable(self.lu, remote_node)
1962

    
1963
        old_node_info = self.cfg.GetNodeInfo(secondary_node)
1964
        assert old_node_info is not None
1965
        if old_node_info.offline and not self.early_release:
1966
          # doesn't make sense to delay the release
1967
          self.early_release = True
1968
          self.lu.LogInfo("Old secondary %s is offline, automatically enabling"
1969
                          " early-release mode", secondary_node)
1970

    
1971
      else:
1972
        raise errors.ProgrammerError("Unhandled disk replace mode (%s)" %
1973
                                     self.mode)
1974

    
1975
      # If not specified all disks should be replaced
1976
      if not self.disks:
1977
        self.disks = range(len(self.instance.disks))
1978

    
1979
    # TODO: This is ugly, but right now we can't distinguish between internal
1980
    # submitted opcode and external one. We should fix that.
1981
    if self.remote_node_info:
1982
      # We change the node, lets verify it still meets instance policy
1983
      new_group_info = self.cfg.GetNodeGroup(self.remote_node_info.group)
1984
      cluster = self.cfg.GetClusterInfo()
1985
      ipolicy = ganeti.masterd.instance.CalculateGroupIPolicy(cluster,
1986
                                                              new_group_info)
1987
      CheckTargetNodeIPolicy(self, ipolicy, instance, self.remote_node_info,
1988
                             self.cfg, ignore=self.ignore_ipolicy)
1989

    
1990
    for node in check_nodes:
1991
      CheckNodeOnline(self.lu, node)
1992

    
1993
    touched_nodes = frozenset(node_name for node_name in [self.new_node,
1994
                                                          self.other_node,
1995
                                                          self.target_node]
1996
                              if node_name is not None)
1997

    
1998
    # Release unneeded node and node resource locks
1999
    ReleaseLocks(self.lu, locking.LEVEL_NODE, keep=touched_nodes)
2000
    ReleaseLocks(self.lu, locking.LEVEL_NODE_RES, keep=touched_nodes)
2001
    ReleaseLocks(self.lu, locking.LEVEL_NODE_ALLOC)
2002

    
2003
    # Release any owned node group
2004
    ReleaseLocks(self.lu, locking.LEVEL_NODEGROUP)
2005

    
2006
    # Check whether disks are valid
2007
    for disk_idx in self.disks:
2008
      instance.FindDisk(disk_idx)
2009

    
2010
    # Get secondary node IP addresses
2011
    self.node_secondary_ip = dict((name, node.secondary_ip) for (name, node)
2012
                                  in self.cfg.GetMultiNodeInfo(touched_nodes))
2013

    
2014
  def Exec(self, feedback_fn):
2015
    """Execute disk replacement.
2016

2017
    This dispatches the disk replacement to the appropriate handler.
2018

2019
    """
2020
    if __debug__:
2021
      # Verify owned locks before starting operation
2022
      owned_nodes = self.lu.owned_locks(locking.LEVEL_NODE)
2023
      assert set(owned_nodes) == set(self.node_secondary_ip), \
2024
          ("Incorrect node locks, owning %s, expected %s" %
2025
           (owned_nodes, self.node_secondary_ip.keys()))
2026
      assert (self.lu.owned_locks(locking.LEVEL_NODE) ==
2027
              self.lu.owned_locks(locking.LEVEL_NODE_RES))
2028
      assert not self.lu.glm.is_owned(locking.LEVEL_NODE_ALLOC)
2029

    
2030
      owned_instances = self.lu.owned_locks(locking.LEVEL_INSTANCE)
2031
      assert list(owned_instances) == [self.instance_name], \
2032
          "Instance '%s' not locked" % self.instance_name
2033

    
2034
      assert not self.lu.glm.is_owned(locking.LEVEL_NODEGROUP), \
2035
          "Should not own any node group lock at this point"
2036

    
2037
    if not self.disks:
2038
      feedback_fn("No disks need replacement for instance '%s'" %
2039
                  self.instance.name)
2040
      return
2041

    
2042
    feedback_fn("Replacing disk(s) %s for instance '%s'" %
2043
                (utils.CommaJoin(self.disks), self.instance.name))
2044
    feedback_fn("Current primary node: %s" % self.instance.primary_node)
2045
    feedback_fn("Current seconary node: %s" %
2046
                utils.CommaJoin(self.instance.secondary_nodes))
2047

    
2048
    activate_disks = not self.instance.disks_active
2049

    
2050
    # Activate the instance disks if we're replacing them on a down instance
2051
    if activate_disks:
2052
      StartInstanceDisks(self.lu, self.instance, True)
2053

    
2054
    try:
2055
      # Should we replace the secondary node?
2056
      if self.new_node is not None:
2057
        fn = self._ExecDrbd8Secondary
2058
      else:
2059
        fn = self._ExecDrbd8DiskOnly
2060

    
2061
      result = fn(feedback_fn)
2062
    finally:
2063
      # Deactivate the instance disks if we're replacing them on a
2064
      # down instance
2065
      if activate_disks:
2066
        _SafeShutdownInstanceDisks(self.lu, self.instance)
2067

    
2068
    assert not self.lu.owned_locks(locking.LEVEL_NODE)
2069

    
2070
    if __debug__:
2071
      # Verify owned locks
2072
      owned_nodes = self.lu.owned_locks(locking.LEVEL_NODE_RES)
2073
      nodes = frozenset(self.node_secondary_ip)
2074
      assert ((self.early_release and not owned_nodes) or
2075
              (not self.early_release and not (set(owned_nodes) - nodes))), \
2076
        ("Not owning the correct locks, early_release=%s, owned=%r,"
2077
         " nodes=%r" % (self.early_release, owned_nodes, nodes))
2078

    
2079
    return result
2080

    
2081
  def _CheckVolumeGroup(self, nodes):
2082
    self.lu.LogInfo("Checking volume groups")
2083

    
2084
    vgname = self.cfg.GetVGName()
2085

    
2086
    # Make sure volume group exists on all involved nodes
2087
    results = self.rpc.call_vg_list(nodes)
2088
    if not results:
2089
      raise errors.OpExecError("Can't list volume groups on the nodes")
2090

    
2091
    for node in nodes:
2092
      res = results[node]
2093
      res.Raise("Error checking node %s" % node)
2094
      if vgname not in res.payload:
2095
        raise errors.OpExecError("Volume group '%s' not found on node %s" %
2096
                                 (vgname, node))
2097

    
2098
  def _CheckDisksExistence(self, nodes):
2099
    # Check disk existence
2100
    for idx, dev in enumerate(self.instance.disks):
2101
      if idx not in self.disks:
2102
        continue
2103

    
2104
      for node in nodes:
2105
        self.lu.LogInfo("Checking disk/%d on %s", idx, node)
2106
        self.cfg.SetDiskID(dev, node)
2107

    
2108
        result = _BlockdevFind(self, node, dev, self.instance)
2109

    
2110
        msg = result.fail_msg
2111
        if msg or not result.payload:
2112
          if not msg:
2113
            msg = "disk not found"
2114
          raise errors.OpExecError("Can't find disk/%d on node %s: %s" %
2115
                                   (idx, node, msg))
2116

    
2117
  def _CheckDisksConsistency(self, node_name, on_primary, ldisk):
2118
    for idx, dev in enumerate(self.instance.disks):
2119
      if idx not in self.disks:
2120
        continue
2121

    
2122
      self.lu.LogInfo("Checking disk/%d consistency on node %s" %
2123
                      (idx, node_name))
2124

    
2125
      if not CheckDiskConsistency(self.lu, self.instance, dev, node_name,
2126
                                  on_primary, ldisk=ldisk):
2127
        raise errors.OpExecError("Node %s has degraded storage, unsafe to"
2128
                                 " replace disks for instance %s" %
2129
                                 (node_name, self.instance.name))
2130

    
2131
  def _CreateNewStorage(self, node_name):
2132
    """Create new storage on the primary or secondary node.
2133

2134
    This is only used for same-node replaces, not for changing the
2135
    secondary node, hence we don't want to modify the existing disk.
2136

2137
    """
2138
    iv_names = {}
2139

    
2140
    disks = AnnotateDiskParams(self.instance, self.instance.disks, self.cfg)
2141
    for idx, dev in enumerate(disks):
2142
      if idx not in self.disks:
2143
        continue
2144

    
2145
      self.lu.LogInfo("Adding storage on %s for disk/%d", node_name, idx)
2146

    
2147
      self.cfg.SetDiskID(dev, node_name)
2148

    
2149
      lv_names = [".disk%d_%s" % (idx, suffix) for suffix in ["data", "meta"]]
2150
      names = _GenerateUniqueNames(self.lu, lv_names)
2151

    
2152
      (data_disk, meta_disk) = dev.children
2153
      vg_data = data_disk.logical_id[0]
2154
      lv_data = objects.Disk(dev_type=constants.LD_LV, size=dev.size,
2155
                             logical_id=(vg_data, names[0]),
2156
                             params=data_disk.params)
2157
      vg_meta = meta_disk.logical_id[0]
2158
      lv_meta = objects.Disk(dev_type=constants.LD_LV,
2159
                             size=constants.DRBD_META_SIZE,
2160
                             logical_id=(vg_meta, names[1]),
2161
                             params=meta_disk.params)
2162

    
2163
      new_lvs = [lv_data, lv_meta]
2164
      old_lvs = [child.Copy() for child in dev.children]
2165
      iv_names[dev.iv_name] = (dev, old_lvs, new_lvs)
2166
      excl_stor = IsExclusiveStorageEnabledNodeName(self.lu.cfg, node_name)
2167

    
2168
      # we pass force_create=True to force the LVM creation
2169
      for new_lv in new_lvs:
2170
        try:
2171
          _CreateBlockDevInner(self.lu, node_name, self.instance, new_lv, True,
2172
                               GetInstanceInfoText(self.instance), False,
2173
                               excl_stor)
2174
        except errors.DeviceCreationError, e:
2175
          raise errors.OpExecError("Can't create block device: %s" % e.message)
2176

    
2177
    return iv_names
2178

    
2179
  def _CheckDevices(self, node_name, iv_names):
2180
    for name, (dev, _, _) in iv_names.iteritems():
2181
      self.cfg.SetDiskID(dev, node_name)
2182

    
2183
      result = _BlockdevFind(self, node_name, dev, self.instance)
2184

    
2185
      msg = result.fail_msg
2186
      if msg or not result.payload:
2187
        if not msg:
2188
          msg = "disk not found"
2189
        raise errors.OpExecError("Can't find DRBD device %s: %s" %
2190
                                 (name, msg))
2191

    
2192
      if result.payload.is_degraded:
2193
        raise errors.OpExecError("DRBD device %s is degraded!" % name)
2194

    
2195
  def _RemoveOldStorage(self, node_name, iv_names):
2196
    for name, (_, old_lvs, _) in iv_names.iteritems():
2197
      self.lu.LogInfo("Remove logical volumes for %s", name)
2198

    
2199
      for lv in old_lvs:
2200
        self.cfg.SetDiskID(lv, node_name)
2201

    
2202
        msg = self.rpc.call_blockdev_remove(node_name, lv).fail_msg
2203
        if msg:
2204
          self.lu.LogWarning("Can't remove old LV: %s", msg,
2205
                             hint="remove unused LVs manually")
2206

    
2207
  def _ExecDrbd8DiskOnly(self, feedback_fn): # pylint: disable=W0613
2208
    """Replace a disk on the primary or secondary for DRBD 8.
2209

2210
    The algorithm for replace is quite complicated:
2211

2212
      1. for each disk to be replaced:
2213

2214
        1. create new LVs on the target node with unique names
2215
        1. detach old LVs from the drbd device
2216
        1. rename old LVs to name_replaced.<time_t>
2217
        1. rename new LVs to old LVs
2218
        1. attach the new LVs (with the old names now) to the drbd device
2219

2220
      1. wait for sync across all devices
2221

2222
      1. for each modified disk:
2223

2224
        1. remove old LVs (which have the name name_replaces.<time_t>)
2225

2226
    Failures are not very well handled.
2227

2228
    """
2229
    steps_total = 6
2230

    
2231
    # Step: check device activation
2232
    self.lu.LogStep(1, steps_total, "Check device existence")
2233
    self._CheckDisksExistence([self.other_node, self.target_node])
2234
    self._CheckVolumeGroup([self.target_node, self.other_node])
2235

    
2236
    # Step: check other node consistency
2237
    self.lu.LogStep(2, steps_total, "Check peer consistency")
2238
    self._CheckDisksConsistency(self.other_node,
2239
                                self.other_node == self.instance.primary_node,
2240
                                False)
2241

    
2242
    # Step: create new storage
2243
    self.lu.LogStep(3, steps_total, "Allocate new storage")
2244
    iv_names = self._CreateNewStorage(self.target_node)
2245

    
2246
    # Step: for each lv, detach+rename*2+attach
2247
    self.lu.LogStep(4, steps_total, "Changing drbd configuration")
2248
    for dev, old_lvs, new_lvs in iv_names.itervalues():
2249
      self.lu.LogInfo("Detaching %s drbd from local storage", dev.iv_name)
2250

    
2251
      result = self.rpc.call_blockdev_removechildren(self.target_node, dev,
2252
                                                     old_lvs)
2253
      result.Raise("Can't detach drbd from local storage on node"
2254
                   " %s for device %s" % (self.target_node, dev.iv_name))
2255
      #dev.children = []
2256
      #cfg.Update(instance)
2257

    
2258
      # ok, we created the new LVs, so now we know we have the needed
2259
      # storage; as such, we proceed on the target node to rename
2260
      # old_lv to _old, and new_lv to old_lv; note that we rename LVs
2261
      # using the assumption that logical_id == physical_id (which in
2262
      # turn is the unique_id on that node)
2263

    
2264
      # FIXME(iustin): use a better name for the replaced LVs
2265
      temp_suffix = int(time.time())
2266
      ren_fn = lambda d, suff: (d.physical_id[0],
2267
                                d.physical_id[1] + "_replaced-%s" % suff)
2268

    
2269
      # Build the rename list based on what LVs exist on the node
2270
      rename_old_to_new = []
2271
      for to_ren in old_lvs:
2272
        result = self.rpc.call_blockdev_find(self.target_node, to_ren)
2273
        if not result.fail_msg and result.payload:
2274
          # device exists
2275
          rename_old_to_new.append((to_ren, ren_fn(to_ren, temp_suffix)))
2276

    
2277
      self.lu.LogInfo("Renaming the old LVs on the target node")
2278
      result = self.rpc.call_blockdev_rename(self.target_node,
2279
                                             rename_old_to_new)
2280
      result.Raise("Can't rename old LVs on node %s" % self.target_node)
2281

    
2282
      # Now we rename the new LVs to the old LVs
2283
      self.lu.LogInfo("Renaming the new LVs on the target node")
2284
      rename_new_to_old = [(new, old.physical_id)
2285
                           for old, new in zip(old_lvs, new_lvs)]
2286
      result = self.rpc.call_blockdev_rename(self.target_node,
2287
                                             rename_new_to_old)
2288
      result.Raise("Can't rename new LVs on node %s" % self.target_node)
2289

    
2290
      # Intermediate steps of in memory modifications
2291
      for old, new in zip(old_lvs, new_lvs):
2292
        new.logical_id = old.logical_id
2293
        self.cfg.SetDiskID(new, self.target_node)
2294

    
2295
      # We need to modify old_lvs so that removal later removes the
2296
      # right LVs, not the newly added ones; note that old_lvs is a
2297
      # copy here
2298
      for disk in old_lvs:
2299
        disk.logical_id = ren_fn(disk, temp_suffix)
2300
        self.cfg.SetDiskID(disk, self.target_node)
2301

    
2302
      # Now that the new lvs have the old name, we can add them to the device
2303
      self.lu.LogInfo("Adding new mirror component on %s", self.target_node)
2304
      result = self.rpc.call_blockdev_addchildren(self.target_node,
2305
                                                  (dev, self.instance), new_lvs)
2306
      msg = result.fail_msg
2307
      if msg:
2308
        for new_lv in new_lvs:
2309
          msg2 = self.rpc.call_blockdev_remove(self.target_node,
2310
                                               new_lv).fail_msg
2311
          if msg2:
2312
            self.lu.LogWarning("Can't rollback device %s: %s", dev, msg2,
2313
                               hint=("cleanup manually the unused logical"
2314
                                     "volumes"))
2315
        raise errors.OpExecError("Can't add local storage to drbd: %s" % msg)
2316

    
2317
    cstep = itertools.count(5)
2318

    
2319
    if self.early_release:
2320
      self.lu.LogStep(cstep.next(), steps_total, "Removing old storage")
2321
      self._RemoveOldStorage(self.target_node, iv_names)
2322
      # TODO: Check if releasing locks early still makes sense
2323
      ReleaseLocks(self.lu, locking.LEVEL_NODE_RES)
2324
    else:
2325
      # Release all resource locks except those used by the instance
2326
      ReleaseLocks(self.lu, locking.LEVEL_NODE_RES,
2327
                   keep=self.node_secondary_ip.keys())
2328

    
2329
    # Release all node locks while waiting for sync
2330
    ReleaseLocks(self.lu, locking.LEVEL_NODE)
2331

    
2332
    # TODO: Can the instance lock be downgraded here? Take the optional disk
2333
    # shutdown in the caller into consideration.
2334

    
2335
    # Wait for sync
2336
    # This can fail as the old devices are degraded and _WaitForSync
2337
    # does a combined result over all disks, so we don't check its return value
2338
    self.lu.LogStep(cstep.next(), steps_total, "Sync devices")
2339
    WaitForSync(self.lu, self.instance)
2340

    
2341
    # Check all devices manually
2342
    self._CheckDevices(self.instance.primary_node, iv_names)
2343

    
2344
    # Step: remove old storage
2345
    if not self.early_release:
2346
      self.lu.LogStep(cstep.next(), steps_total, "Removing old storage")
2347
      self._RemoveOldStorage(self.target_node, iv_names)
2348

    
2349
  def _ExecDrbd8Secondary(self, feedback_fn):
2350
    """Replace the secondary node for DRBD 8.
2351

2352
    The algorithm for replace is quite complicated:
2353
      - for all disks of the instance:
2354
        - create new LVs on the new node with same names
2355
        - shutdown the drbd device on the old secondary
2356
        - disconnect the drbd network on the primary
2357
        - create the drbd device on the new secondary
2358
        - network attach the drbd on the primary, using an artifice:
2359
          the drbd code for Attach() will connect to the network if it
2360
          finds a device which is connected to the good local disks but
2361
          not network enabled
2362
      - wait for sync across all devices
2363
      - remove all disks from the old secondary
2364

2365
    Failures are not very well handled.
2366

2367
    """
2368
    steps_total = 6
2369

    
2370
    pnode = self.instance.primary_node
2371

    
2372
    # Step: check device activation
2373
    self.lu.LogStep(1, steps_total, "Check device existence")
2374
    self._CheckDisksExistence([self.instance.primary_node])
2375
    self._CheckVolumeGroup([self.instance.primary_node])
2376

    
2377
    # Step: check other node consistency
2378
    self.lu.LogStep(2, steps_total, "Check peer consistency")
2379
    self._CheckDisksConsistency(self.instance.primary_node, True, True)
2380

    
2381
    # Step: create new storage
2382
    self.lu.LogStep(3, steps_total, "Allocate new storage")
2383
    disks = AnnotateDiskParams(self.instance, self.instance.disks, self.cfg)
2384
    excl_stor = IsExclusiveStorageEnabledNodeName(self.lu.cfg, self.new_node)
2385
    for idx, dev in enumerate(disks):
2386
      self.lu.LogInfo("Adding new local storage on %s for disk/%d" %
2387
                      (self.new_node, idx))
2388
      # we pass force_create=True to force LVM creation
2389
      for new_lv in dev.children:
2390
        try:
2391
          _CreateBlockDevInner(self.lu, self.new_node, self.instance, new_lv,
2392
                               True, GetInstanceInfoText(self.instance), False,
2393
                               excl_stor)
2394
        except errors.DeviceCreationError, e:
2395
          raise errors.OpExecError("Can't create block device: %s" % e.message)
2396

    
2397
    # Step 4: dbrd minors and drbd setups changes
2398
    # after this, we must manually remove the drbd minors on both the
2399
    # error and the success paths
2400
    self.lu.LogStep(4, steps_total, "Changing drbd configuration")
2401
    minors = self.cfg.AllocateDRBDMinor([self.new_node
2402
                                         for dev in self.instance.disks],
2403
                                        self.instance.name)
2404
    logging.debug("Allocated minors %r", minors)
2405

    
2406
    iv_names = {}
2407
    for idx, (dev, new_minor) in enumerate(zip(self.instance.disks, minors)):
2408
      self.lu.LogInfo("activating a new drbd on %s for disk/%d" %
2409
                      (self.new_node, idx))
2410
      # create new devices on new_node; note that we create two IDs:
2411
      # one without port, so the drbd will be activated without
2412
      # networking information on the new node at this stage, and one
2413
      # with network, for the latter activation in step 4
2414
      (o_node1, o_node2, o_port, o_minor1, o_minor2, o_secret) = dev.logical_id
2415
      if self.instance.primary_node == o_node1:
2416
        p_minor = o_minor1
2417
      else:
2418
        assert self.instance.primary_node == o_node2, "Three-node instance?"
2419
        p_minor = o_minor2
2420

    
2421
      new_alone_id = (self.instance.primary_node, self.new_node, None,
2422
                      p_minor, new_minor, o_secret)
2423
      new_net_id = (self.instance.primary_node, self.new_node, o_port,
2424
                    p_minor, new_minor, o_secret)
2425

    
2426
      iv_names[idx] = (dev, dev.children, new_net_id)
2427
      logging.debug("Allocated new_minor: %s, new_logical_id: %s", new_minor,
2428
                    new_net_id)
2429
      new_drbd = objects.Disk(dev_type=constants.LD_DRBD8,
2430
                              logical_id=new_alone_id,
2431
                              children=dev.children,
2432
                              size=dev.size,
2433
                              params={})
2434
      (anno_new_drbd,) = AnnotateDiskParams(self.instance, [new_drbd],
2435
                                            self.cfg)
2436
      try:
2437
        CreateSingleBlockDev(self.lu, self.new_node, self.instance,
2438
                             anno_new_drbd,
2439
                             GetInstanceInfoText(self.instance), False,
2440
                             excl_stor)
2441
      except errors.GenericError:
2442
        self.cfg.ReleaseDRBDMinors(self.instance.name)
2443
        raise
2444

    
2445
    # We have new devices, shutdown the drbd on the old secondary
2446
    for idx, dev in enumerate(self.instance.disks):
2447
      self.lu.LogInfo("Shutting down drbd for disk/%d on old node", idx)
2448
      self.cfg.SetDiskID(dev, self.target_node)
2449
      msg = self.rpc.call_blockdev_shutdown(self.target_node,
2450
                                            (dev, self.instance)).fail_msg
2451
      if msg:
2452
        self.lu.LogWarning("Failed to shutdown drbd for disk/%d on old"
2453
                           "node: %s" % (idx, msg),
2454
                           hint=("Please cleanup this device manually as"
2455
                                 " soon as possible"))
2456

    
2457
    self.lu.LogInfo("Detaching primary drbds from the network (=> standalone)")
2458
    result = self.rpc.call_drbd_disconnect_net([pnode], self.node_secondary_ip,
2459
                                               self.instance.disks)[pnode]
2460

    
2461
    msg = result.fail_msg
2462
    if msg:
2463
      # detaches didn't succeed (unlikely)
2464
      self.cfg.ReleaseDRBDMinors(self.instance.name)
2465
      raise errors.OpExecError("Can't detach the disks from the network on"
2466
                               " old node: %s" % (msg,))
2467

    
2468
    # if we managed to detach at least one, we update all the disks of
2469
    # the instance to point to the new secondary
2470
    self.lu.LogInfo("Updating instance configuration")
2471
    for dev, _, new_logical_id in iv_names.itervalues():
2472
      dev.logical_id = new_logical_id
2473
      self.cfg.SetDiskID(dev, self.instance.primary_node)
2474

    
2475
    self.cfg.Update(self.instance, feedback_fn)
2476

    
2477
    # Release all node locks (the configuration has been updated)
2478
    ReleaseLocks(self.lu, locking.LEVEL_NODE)
2479

    
2480
    # and now perform the drbd attach
2481
    self.lu.LogInfo("Attaching primary drbds to new secondary"
2482
                    " (standalone => connected)")
2483
    result = self.rpc.call_drbd_attach_net([self.instance.primary_node,
2484
                                            self.new_node],
2485
                                           self.node_secondary_ip,
2486
                                           (self.instance.disks, self.instance),
2487
                                           self.instance.name,
2488
                                           False)
2489
    for to_node, to_result in result.items():
2490
      msg = to_result.fail_msg
2491
      if msg:
2492
        self.lu.LogWarning("Can't attach drbd disks on node %s: %s",
2493
                           to_node, msg,
2494
                           hint=("please do a gnt-instance info to see the"
2495
                                 " status of disks"))
2496

    
2497
    cstep = itertools.count(5)
2498

    
2499
    if self.early_release:
2500
      self.lu.LogStep(cstep.next(), steps_total, "Removing old storage")
2501
      self._RemoveOldStorage(self.target_node, iv_names)
2502
      # TODO: Check if releasing locks early still makes sense
2503
      ReleaseLocks(self.lu, locking.LEVEL_NODE_RES)
2504
    else:
2505
      # Release all resource locks except those used by the instance
2506
      ReleaseLocks(self.lu, locking.LEVEL_NODE_RES,
2507
                   keep=self.node_secondary_ip.keys())
2508

    
2509
    # TODO: Can the instance lock be downgraded here? Take the optional disk
2510
    # shutdown in the caller into consideration.
2511

    
2512
    # Wait for sync
2513
    # This can fail as the old devices are degraded and _WaitForSync
2514
    # does a combined result over all disks, so we don't check its return value
2515
    self.lu.LogStep(cstep.next(), steps_total, "Sync devices")
2516
    WaitForSync(self.lu, self.instance)
2517

    
2518
    # Check all devices manually
2519
    self._CheckDevices(self.instance.primary_node, iv_names)
2520

    
2521
    # Step: remove old storage
2522
    if not self.early_release:
2523
      self.lu.LogStep(cstep.next(), steps_total, "Removing old storage")
2524
      self._RemoveOldStorage(self.target_node, iv_names)