6478e51a941782c42d103b97623013cbb80a1f0c
[ganeti-github.git] / lib / cmdlib / instance_storage.py
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)