Statistics
| Branch: | Tag: | Revision:

root / lib / cmdlib / instance.py @ b4076d8a

History | View | Annotate | Download (140.6 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 instances."""
23

    
24
import OpenSSL
25
import copy
26
import logging
27
import os
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 hypervisor
34
from ganeti import locking
35
from ganeti.masterd import iallocator
36
from ganeti import masterd
37
from ganeti import netutils
38
from ganeti import objects
39
from ganeti import opcodes
40
from ganeti import pathutils
41
from ganeti import rpc
42
from ganeti import utils
43

    
44
from ganeti.cmdlib.base import NoHooksLU, LogicalUnit, ResultWithJobs
45

    
46
from ganeti.cmdlib.common import INSTANCE_DOWN, \
47
  INSTANCE_NOT_RUNNING, CAN_CHANGE_INSTANCE_OFFLINE, CheckNodeOnline, \
48
  ShareAll, GetDefaultIAllocator, CheckInstanceNodeGroups, \
49
  LoadNodeEvacResult, CheckIAllocatorOrNode, CheckParamsNotGlobal, \
50
  IsExclusiveStorageEnabledNode, CheckHVParams, CheckOSParams, \
51
  AnnotateDiskParams, GetUpdatedParams, ExpandInstanceName, \
52
  ComputeIPolicySpecViolation, CheckInstanceState, ExpandNodeName
53
from ganeti.cmdlib.instance_storage import CreateDisks, \
54
  CheckNodesFreeDiskPerVG, WipeDisks, WipeOrCleanupDisks, WaitForSync, \
55
  IsExclusiveStorageEnabledNodeName, CreateSingleBlockDev, ComputeDisks, \
56
  CheckRADOSFreeSpace, ComputeDiskSizePerVG, GenerateDiskTemplate, \
57
  StartInstanceDisks, ShutdownInstanceDisks, AssembleInstanceDisks
58
from ganeti.cmdlib.instance_utils import BuildInstanceHookEnvByObject, \
59
  GetClusterDomainSecret, BuildInstanceHookEnv, NICListToTuple, \
60
  NICToTuple, CheckNodeNotDrained, RemoveInstance, CopyLockList, \
61
  ReleaseLocks, CheckNodeVmCapable, CheckTargetNodeIPolicy, \
62
  GetInstanceInfoText, RemoveDisks, CheckNodeFreeMemory, \
63
  CheckInstanceBridgesExist, CheckNicsBridgesExist, CheckNodeHasOS
64

    
65
import ganeti.masterd.instance
66

    
67

    
68
#: Type description for changes as returned by L{_ApplyContainerMods}'s
69
#: callbacks
70
_TApplyContModsCbChanges = \
71
  ht.TMaybeListOf(ht.TAnd(ht.TIsLength(2), ht.TItems([
72
    ht.TNonEmptyString,
73
    ht.TAny,
74
    ])))
75

    
76

    
77
def _CheckHostnameSane(lu, name):
78
  """Ensures that a given hostname resolves to a 'sane' name.
79

80
  The given name is required to be a prefix of the resolved hostname,
81
  to prevent accidental mismatches.
82

83
  @param lu: the logical unit on behalf of which we're checking
84
  @param name: the name we should resolve and check
85
  @return: the resolved hostname object
86

87
  """
88
  hostname = netutils.GetHostname(name=name)
89
  if hostname.name != name:
90
    lu.LogInfo("Resolved given name '%s' to '%s'", name, hostname.name)
91
  if not utils.MatchNameComponent(name, [hostname.name]):
92
    raise errors.OpPrereqError(("Resolved hostname '%s' does not look the"
93
                                " same as given hostname '%s'") %
94
                               (hostname.name, name), errors.ECODE_INVAL)
95
  return hostname
96

    
97

    
98
def _CheckOpportunisticLocking(op):
99
  """Generate error if opportunistic locking is not possible.
100

101
  """
102
  if op.opportunistic_locking and not op.iallocator:
103
    raise errors.OpPrereqError("Opportunistic locking is only available in"
104
                               " combination with an instance allocator",
105
                               errors.ECODE_INVAL)
106

    
107

    
108
def _CreateInstanceAllocRequest(op, disks, nics, beparams, node_whitelist):
109
  """Wrapper around IAReqInstanceAlloc.
110

111
  @param op: The instance opcode
112
  @param disks: The computed disks
113
  @param nics: The computed nics
114
  @param beparams: The full filled beparams
115
  @param node_whitelist: List of nodes which should appear as online to the
116
    allocator (unless the node is already marked offline)
117

118
  @returns: A filled L{iallocator.IAReqInstanceAlloc}
119

120
  """
121
  spindle_use = beparams[constants.BE_SPINDLE_USE]
122
  return iallocator.IAReqInstanceAlloc(name=op.instance_name,
123
                                       disk_template=op.disk_template,
124
                                       tags=op.tags,
125
                                       os=op.os_type,
126
                                       vcpus=beparams[constants.BE_VCPUS],
127
                                       memory=beparams[constants.BE_MAXMEM],
128
                                       spindle_use=spindle_use,
129
                                       disks=disks,
130
                                       nics=[n.ToDict() for n in nics],
131
                                       hypervisor=op.hypervisor,
132
                                       node_whitelist=node_whitelist)
133

    
134

    
135
def _ComputeFullBeParams(op, cluster):
136
  """Computes the full beparams.
137

138
  @param op: The instance opcode
139
  @param cluster: The cluster config object
140

141
  @return: The fully filled beparams
142

143
  """
144
  default_beparams = cluster.beparams[constants.PP_DEFAULT]
145
  for param, value in op.beparams.iteritems():
146
    if value == constants.VALUE_AUTO:
147
      op.beparams[param] = default_beparams[param]
148
  objects.UpgradeBeParams(op.beparams)
149
  utils.ForceDictType(op.beparams, constants.BES_PARAMETER_TYPES)
150
  return cluster.SimpleFillBE(op.beparams)
151

    
152

    
153
def _ComputeNics(op, cluster, default_ip, cfg, ec_id):
154
  """Computes the nics.
155

156
  @param op: The instance opcode
157
  @param cluster: Cluster configuration object
158
  @param default_ip: The default ip to assign
159
  @param cfg: An instance of the configuration object
160
  @param ec_id: Execution context ID
161

162
  @returns: The build up nics
163

164
  """
165
  nics = []
166
  for nic in op.nics:
167
    nic_mode_req = nic.get(constants.INIC_MODE, None)
168
    nic_mode = nic_mode_req
169
    if nic_mode is None or nic_mode == constants.VALUE_AUTO:
170
      nic_mode = cluster.nicparams[constants.PP_DEFAULT][constants.NIC_MODE]
171

    
172
    net = nic.get(constants.INIC_NETWORK, None)
173
    link = nic.get(constants.NIC_LINK, None)
174
    ip = nic.get(constants.INIC_IP, None)
175

    
176
    if net is None or net.lower() == constants.VALUE_NONE:
177
      net = None
178
    else:
179
      if nic_mode_req is not None or link is not None:
180
        raise errors.OpPrereqError("If network is given, no mode or link"
181
                                   " is allowed to be passed",
182
                                   errors.ECODE_INVAL)
183

    
184
    # ip validity checks
185
    if ip is None or ip.lower() == constants.VALUE_NONE:
186
      nic_ip = None
187
    elif ip.lower() == constants.VALUE_AUTO:
188
      if not op.name_check:
189
        raise errors.OpPrereqError("IP address set to auto but name checks"
190
                                   " have been skipped",
191
                                   errors.ECODE_INVAL)
192
      nic_ip = default_ip
193
    else:
194
      # We defer pool operations until later, so that the iallocator has
195
      # filled in the instance's node(s) dimara
196
      if ip.lower() == constants.NIC_IP_POOL:
197
        if net is None:
198
          raise errors.OpPrereqError("if ip=pool, parameter network"
199
                                     " must be passed too",
200
                                     errors.ECODE_INVAL)
201

    
202
      elif not netutils.IPAddress.IsValid(ip):
203
        raise errors.OpPrereqError("Invalid IP address '%s'" % ip,
204
                                   errors.ECODE_INVAL)
205

    
206
      nic_ip = ip
207

    
208
    # TODO: check the ip address for uniqueness
209
    if nic_mode == constants.NIC_MODE_ROUTED and not nic_ip:
210
      raise errors.OpPrereqError("Routed nic mode requires an ip address",
211
                                 errors.ECODE_INVAL)
212

    
213
    # MAC address verification
214
    mac = nic.get(constants.INIC_MAC, constants.VALUE_AUTO)
215
    if mac not in (constants.VALUE_AUTO, constants.VALUE_GENERATE):
216
      mac = utils.NormalizeAndValidateMac(mac)
217

    
218
      try:
219
        # TODO: We need to factor this out
220
        cfg.ReserveMAC(mac, ec_id)
221
      except errors.ReservationError:
222
        raise errors.OpPrereqError("MAC address %s already in use"
223
                                   " in cluster" % mac,
224
                                   errors.ECODE_NOTUNIQUE)
225

    
226
    #  Build nic parameters
227
    nicparams = {}
228
    if nic_mode_req:
229
      nicparams[constants.NIC_MODE] = nic_mode
230
    if link:
231
      nicparams[constants.NIC_LINK] = link
232

    
233
    check_params = cluster.SimpleFillNIC(nicparams)
234
    objects.NIC.CheckParameterSyntax(check_params)
235
    net_uuid = cfg.LookupNetwork(net)
236
    name = nic.get(constants.INIC_NAME, None)
237
    if name is not None and name.lower() == constants.VALUE_NONE:
238
      name = None
239
    nic_obj = objects.NIC(mac=mac, ip=nic_ip, name=name,
240
                          network=net_uuid, nicparams=nicparams)
241
    nic_obj.uuid = cfg.GenerateUniqueID(ec_id)
242
    nics.append(nic_obj)
243

    
244
  return nics
245

    
246

    
247
def _CheckForConflictingIp(lu, ip, node):
248
  """In case of conflicting IP address raise error.
249

250
  @type ip: string
251
  @param ip: IP address
252
  @type node: string
253
  @param node: node name
254

255
  """
256
  (conf_net, _) = lu.cfg.CheckIPInNodeGroup(ip, node)
257
  if conf_net is not None:
258
    raise errors.OpPrereqError(("The requested IP address (%s) belongs to"
259
                                " network %s, but the target NIC does not." %
260
                                (ip, conf_net)),
261
                               errors.ECODE_STATE)
262

    
263
  return (None, None)
264

    
265

    
266
def _ComputeIPolicyInstanceSpecViolation(
267
  ipolicy, instance_spec, disk_template,
268
  _compute_fn=ComputeIPolicySpecViolation):
269
  """Compute if instance specs meets the specs of ipolicy.
270

271
  @type ipolicy: dict
272
  @param ipolicy: The ipolicy to verify against
273
  @param instance_spec: dict
274
  @param instance_spec: The instance spec to verify
275
  @type disk_template: string
276
  @param disk_template: the disk template of the instance
277
  @param _compute_fn: The function to verify ipolicy (unittest only)
278
  @see: L{ComputeIPolicySpecViolation}
279

280
  """
281
  mem_size = instance_spec.get(constants.ISPEC_MEM_SIZE, None)
282
  cpu_count = instance_spec.get(constants.ISPEC_CPU_COUNT, None)
283
  disk_count = instance_spec.get(constants.ISPEC_DISK_COUNT, 0)
284
  disk_sizes = instance_spec.get(constants.ISPEC_DISK_SIZE, [])
285
  nic_count = instance_spec.get(constants.ISPEC_NIC_COUNT, 0)
286
  spindle_use = instance_spec.get(constants.ISPEC_SPINDLE_USE, None)
287

    
288
  return _compute_fn(ipolicy, mem_size, cpu_count, disk_count, nic_count,
289
                     disk_sizes, spindle_use, disk_template)
290

    
291

    
292
def _CheckOSVariant(os_obj, name):
293
  """Check whether an OS name conforms to the os variants specification.
294

295
  @type os_obj: L{objects.OS}
296
  @param os_obj: OS object to check
297
  @type name: string
298
  @param name: OS name passed by the user, to check for validity
299

300
  """
301
  variant = objects.OS.GetVariant(name)
302
  if not os_obj.supported_variants:
303
    if variant:
304
      raise errors.OpPrereqError("OS '%s' doesn't support variants ('%s'"
305
                                 " passed)" % (os_obj.name, variant),
306
                                 errors.ECODE_INVAL)
307
    return
308
  if not variant:
309
    raise errors.OpPrereqError("OS name must include a variant",
310
                               errors.ECODE_INVAL)
311

    
312
  if variant not in os_obj.supported_variants:
313
    raise errors.OpPrereqError("Unsupported OS variant", errors.ECODE_INVAL)
314

    
315

    
316
class LUInstanceCreate(LogicalUnit):
317
  """Create an instance.
318

319
  """
320
  HPATH = "instance-add"
321
  HTYPE = constants.HTYPE_INSTANCE
322
  REQ_BGL = False
323

    
324
  def CheckArguments(self):
325
    """Check arguments.
326

327
    """
328
    # do not require name_check to ease forward/backward compatibility
329
    # for tools
330
    if self.op.no_install and self.op.start:
331
      self.LogInfo("No-installation mode selected, disabling startup")
332
      self.op.start = False
333
    # validate/normalize the instance name
334
    self.op.instance_name = \
335
      netutils.Hostname.GetNormalizedName(self.op.instance_name)
336

    
337
    if self.op.ip_check and not self.op.name_check:
338
      # TODO: make the ip check more flexible and not depend on the name check
339
      raise errors.OpPrereqError("Cannot do IP address check without a name"
340
                                 " check", errors.ECODE_INVAL)
341

    
342
    # check nics' parameter names
343
    for nic in self.op.nics:
344
      utils.ForceDictType(nic, constants.INIC_PARAMS_TYPES)
345
    # check that NIC's parameters names are unique and valid
346
    utils.ValidateDeviceNames("NIC", self.op.nics)
347

    
348
    # check that disk's names are unique and valid
349
    utils.ValidateDeviceNames("disk", self.op.disks)
350

    
351
    cluster = self.cfg.GetClusterInfo()
352
    if not self.op.disk_template in cluster.enabled_disk_templates:
353
      raise errors.OpPrereqError("Cannot create an instance with disk template"
354
                                 " '%s', because it is not enabled in the"
355
                                 " cluster. Enabled disk templates are: %s." %
356
                                 (self.op.disk_template,
357
                                  ",".join(cluster.enabled_disk_templates)))
358

    
359
    # check disks. parameter names and consistent adopt/no-adopt strategy
360
    has_adopt = has_no_adopt = False
361
    for disk in self.op.disks:
362
      if self.op.disk_template != constants.DT_EXT:
363
        utils.ForceDictType(disk, constants.IDISK_PARAMS_TYPES)
364
      if constants.IDISK_ADOPT in disk:
365
        has_adopt = True
366
      else:
367
        has_no_adopt = True
368
    if has_adopt and has_no_adopt:
369
      raise errors.OpPrereqError("Either all disks are adopted or none is",
370
                                 errors.ECODE_INVAL)
371
    if has_adopt:
372
      if self.op.disk_template not in constants.DTS_MAY_ADOPT:
373
        raise errors.OpPrereqError("Disk adoption is not supported for the"
374
                                   " '%s' disk template" %
375
                                   self.op.disk_template,
376
                                   errors.ECODE_INVAL)
377
      if self.op.iallocator is not None:
378
        raise errors.OpPrereqError("Disk adoption not allowed with an"
379
                                   " iallocator script", errors.ECODE_INVAL)
380
      if self.op.mode == constants.INSTANCE_IMPORT:
381
        raise errors.OpPrereqError("Disk adoption not allowed for"
382
                                   " instance import", errors.ECODE_INVAL)
383
    else:
384
      if self.op.disk_template in constants.DTS_MUST_ADOPT:
385
        raise errors.OpPrereqError("Disk template %s requires disk adoption,"
386
                                   " but no 'adopt' parameter given" %
387
                                   self.op.disk_template,
388
                                   errors.ECODE_INVAL)
389

    
390
    self.adopt_disks = has_adopt
391

    
392
    # instance name verification
393
    if self.op.name_check:
394
      self.hostname1 = _CheckHostnameSane(self, self.op.instance_name)
395
      self.op.instance_name = self.hostname1.name
396
      # used in CheckPrereq for ip ping check
397
      self.check_ip = self.hostname1.ip
398
    else:
399
      self.check_ip = None
400

    
401
    # file storage checks
402
    if (self.op.file_driver and
403
        not self.op.file_driver in constants.FILE_DRIVER):
404
      raise errors.OpPrereqError("Invalid file driver name '%s'" %
405
                                 self.op.file_driver, errors.ECODE_INVAL)
406

    
407
    # set default file_driver if unset and required
408
    if (not self.op.file_driver and
409
        self.op.disk_template in [constants.DT_FILE,
410
                                  constants.DT_SHARED_FILE]):
411
      self.op.file_driver = constants.FD_LOOP
412

    
413
    if self.op.disk_template == constants.DT_FILE:
414
      opcodes.RequireFileStorage()
415
    elif self.op.disk_template == constants.DT_SHARED_FILE:
416
      opcodes.RequireSharedFileStorage()
417

    
418
    ### Node/iallocator related checks
419
    CheckIAllocatorOrNode(self, "iallocator", "pnode")
420

    
421
    if self.op.pnode is not None:
422
      if self.op.disk_template in constants.DTS_INT_MIRROR:
423
        if self.op.snode is None:
424
          raise errors.OpPrereqError("The networked disk templates need"
425
                                     " a mirror node", errors.ECODE_INVAL)
426
      elif self.op.snode:
427
        self.LogWarning("Secondary node will be ignored on non-mirrored disk"
428
                        " template")
429
        self.op.snode = None
430

    
431
    _CheckOpportunisticLocking(self.op)
432

    
433
    self._cds = GetClusterDomainSecret()
434

    
435
    if self.op.mode == constants.INSTANCE_IMPORT:
436
      # On import force_variant must be True, because if we forced it at
437
      # initial install, our only chance when importing it back is that it
438
      # works again!
439
      self.op.force_variant = True
440

    
441
      if self.op.no_install:
442
        self.LogInfo("No-installation mode has no effect during import")
443

    
444
    elif self.op.mode == constants.INSTANCE_CREATE:
445
      if self.op.os_type is None:
446
        raise errors.OpPrereqError("No guest OS specified",
447
                                   errors.ECODE_INVAL)
448
      if self.op.os_type in self.cfg.GetClusterInfo().blacklisted_os:
449
        raise errors.OpPrereqError("Guest OS '%s' is not allowed for"
450
                                   " installation" % self.op.os_type,
451
                                   errors.ECODE_STATE)
452
      if self.op.disk_template is None:
453
        raise errors.OpPrereqError("No disk template specified",
454
                                   errors.ECODE_INVAL)
455

    
456
    elif self.op.mode == constants.INSTANCE_REMOTE_IMPORT:
457
      # Check handshake to ensure both clusters have the same domain secret
458
      src_handshake = self.op.source_handshake
459
      if not src_handshake:
460
        raise errors.OpPrereqError("Missing source handshake",
461
                                   errors.ECODE_INVAL)
462

    
463
      errmsg = masterd.instance.CheckRemoteExportHandshake(self._cds,
464
                                                           src_handshake)
465
      if errmsg:
466
        raise errors.OpPrereqError("Invalid handshake: %s" % errmsg,
467
                                   errors.ECODE_INVAL)
468

    
469
      # Load and check source CA
470
      self.source_x509_ca_pem = self.op.source_x509_ca
471
      if not self.source_x509_ca_pem:
472
        raise errors.OpPrereqError("Missing source X509 CA",
473
                                   errors.ECODE_INVAL)
474

    
475
      try:
476
        (cert, _) = utils.LoadSignedX509Certificate(self.source_x509_ca_pem,
477
                                                    self._cds)
478
      except OpenSSL.crypto.Error, err:
479
        raise errors.OpPrereqError("Unable to load source X509 CA (%s)" %
480
                                   (err, ), errors.ECODE_INVAL)
481

    
482
      (errcode, msg) = utils.VerifyX509Certificate(cert, None, None)
483
      if errcode is not None:
484
        raise errors.OpPrereqError("Invalid source X509 CA (%s)" % (msg, ),
485
                                   errors.ECODE_INVAL)
486

    
487
      self.source_x509_ca = cert
488

    
489
      src_instance_name = self.op.source_instance_name
490
      if not src_instance_name:
491
        raise errors.OpPrereqError("Missing source instance name",
492
                                   errors.ECODE_INVAL)
493

    
494
      self.source_instance_name = \
495
        netutils.GetHostname(name=src_instance_name).name
496

    
497
    else:
498
      raise errors.OpPrereqError("Invalid instance creation mode %r" %
499
                                 self.op.mode, errors.ECODE_INVAL)
500

    
501
  def ExpandNames(self):
502
    """ExpandNames for CreateInstance.
503

504
    Figure out the right locks for instance creation.
505

506
    """
507
    self.needed_locks = {}
508

    
509
    instance_name = self.op.instance_name
510
    # this is just a preventive check, but someone might still add this
511
    # instance in the meantime, and creation will fail at lock-add time
512
    if instance_name in self.cfg.GetInstanceList():
513
      raise errors.OpPrereqError("Instance '%s' is already in the cluster" %
514
                                 instance_name, errors.ECODE_EXISTS)
515

    
516
    self.add_locks[locking.LEVEL_INSTANCE] = instance_name
517

    
518
    if self.op.iallocator:
519
      # TODO: Find a solution to not lock all nodes in the cluster, e.g. by
520
      # specifying a group on instance creation and then selecting nodes from
521
      # that group
522
      self.needed_locks[locking.LEVEL_NODE] = locking.ALL_SET
523
      self.needed_locks[locking.LEVEL_NODE_ALLOC] = locking.ALL_SET
524

    
525
      if self.op.opportunistic_locking:
526
        self.opportunistic_locks[locking.LEVEL_NODE] = True
527
    else:
528
      self.op.pnode = ExpandNodeName(self.cfg, self.op.pnode)
529
      nodelist = [self.op.pnode]
530
      if self.op.snode is not None:
531
        self.op.snode = ExpandNodeName(self.cfg, self.op.snode)
532
        nodelist.append(self.op.snode)
533
      self.needed_locks[locking.LEVEL_NODE] = nodelist
534

    
535
    # in case of import lock the source node too
536
    if self.op.mode == constants.INSTANCE_IMPORT:
537
      src_node = self.op.src_node
538
      src_path = self.op.src_path
539

    
540
      if src_path is None:
541
        self.op.src_path = src_path = self.op.instance_name
542

    
543
      if src_node is None:
544
        self.needed_locks[locking.LEVEL_NODE] = locking.ALL_SET
545
        self.needed_locks[locking.LEVEL_NODE_ALLOC] = locking.ALL_SET
546
        self.op.src_node = None
547
        if os.path.isabs(src_path):
548
          raise errors.OpPrereqError("Importing an instance from a path"
549
                                     " requires a source node option",
550
                                     errors.ECODE_INVAL)
551
      else:
552
        self.op.src_node = src_node = ExpandNodeName(self.cfg, src_node)
553
        if self.needed_locks[locking.LEVEL_NODE] is not locking.ALL_SET:
554
          self.needed_locks[locking.LEVEL_NODE].append(src_node)
555
        if not os.path.isabs(src_path):
556
          self.op.src_path = src_path = \
557
            utils.PathJoin(pathutils.EXPORT_DIR, src_path)
558

    
559
    self.needed_locks[locking.LEVEL_NODE_RES] = \
560
      CopyLockList(self.needed_locks[locking.LEVEL_NODE])
561

    
562
    # Optimistically acquire shared group locks (we're reading the
563
    # configuration).  We can't just call GetInstanceNodeGroups, because the
564
    # instance doesn't exist yet. Therefore we lock all node groups of all
565
    # nodes we have.
566
    if self.needed_locks[locking.LEVEL_NODE] == locking.ALL_SET:
567
      # In the case we lock all nodes for opportunistic allocation, we have no
568
      # choice than to lock all groups, because they're allocated before nodes.
569
      # This is sad, but true. At least we release all those we don't need in
570
      # CheckPrereq later.
571
      self.needed_locks[locking.LEVEL_NODEGROUP] = locking.ALL_SET
572
    else:
573
      self.needed_locks[locking.LEVEL_NODEGROUP] = \
574
        list(self.cfg.GetNodeGroupsFromNodes(
575
          self.needed_locks[locking.LEVEL_NODE]))
576
    self.share_locks[locking.LEVEL_NODEGROUP] = 1
577

    
578
  def DeclareLocks(self, level):
579
    if level == locking.LEVEL_NODE_RES and \
580
      self.opportunistic_locks[locking.LEVEL_NODE]:
581
      # Even when using opportunistic locking, we require the same set of
582
      # NODE_RES locks as we got NODE locks
583
      self.needed_locks[locking.LEVEL_NODE_RES] = \
584
        self.owned_locks(locking.LEVEL_NODE)
585

    
586
  def _RunAllocator(self):
587
    """Run the allocator based on input opcode.
588

589
    """
590
    if self.op.opportunistic_locking:
591
      # Only consider nodes for which a lock is held
592
      node_whitelist = list(self.owned_locks(locking.LEVEL_NODE))
593
    else:
594
      node_whitelist = None
595

    
596
    #TODO Export network to iallocator so that it chooses a pnode
597
    #     in a nodegroup that has the desired network connected to
598
    req = _CreateInstanceAllocRequest(self.op, self.disks,
599
                                      self.nics, self.be_full,
600
                                      node_whitelist)
601
    ial = iallocator.IAllocator(self.cfg, self.rpc, req)
602

    
603
    ial.Run(self.op.iallocator)
604

    
605
    if not ial.success:
606
      # When opportunistic locks are used only a temporary failure is generated
607
      if self.op.opportunistic_locking:
608
        ecode = errors.ECODE_TEMP_NORES
609
      else:
610
        ecode = errors.ECODE_NORES
611

    
612
      raise errors.OpPrereqError("Can't compute nodes using"
613
                                 " iallocator '%s': %s" %
614
                                 (self.op.iallocator, ial.info),
615
                                 ecode)
616

    
617
    self.op.pnode = ial.result[0]
618
    self.LogInfo("Selected nodes for instance %s via iallocator %s: %s",
619
                 self.op.instance_name, self.op.iallocator,
620
                 utils.CommaJoin(ial.result))
621

    
622
    assert req.RequiredNodes() in (1, 2), "Wrong node count from iallocator"
623

    
624
    if req.RequiredNodes() == 2:
625
      self.op.snode = ial.result[1]
626

    
627
  def BuildHooksEnv(self):
628
    """Build hooks env.
629

630
    This runs on master, primary and secondary nodes of the instance.
631

632
    """
633
    env = {
634
      "ADD_MODE": self.op.mode,
635
      }
636
    if self.op.mode == constants.INSTANCE_IMPORT:
637
      env["SRC_NODE"] = self.op.src_node
638
      env["SRC_PATH"] = self.op.src_path
639
      env["SRC_IMAGES"] = self.src_images
640

    
641
    env.update(BuildInstanceHookEnv(
642
      name=self.op.instance_name,
643
      primary_node=self.op.pnode,
644
      secondary_nodes=self.secondaries,
645
      status=self.op.start,
646
      os_type=self.op.os_type,
647
      minmem=self.be_full[constants.BE_MINMEM],
648
      maxmem=self.be_full[constants.BE_MAXMEM],
649
      vcpus=self.be_full[constants.BE_VCPUS],
650
      nics=NICListToTuple(self, self.nics),
651
      disk_template=self.op.disk_template,
652
      disks=[(d[constants.IDISK_NAME], d.get("uuid", ""),
653
              d[constants.IDISK_SIZE], d[constants.IDISK_MODE], {})
654
             for d in self.disks],
655
      bep=self.be_full,
656
      hvp=self.hv_full,
657
      hypervisor_name=self.op.hypervisor,
658
      tags=self.op.tags,
659
      ))
660

    
661
    return env
662

    
663
  def BuildHooksNodes(self):
664
    """Build hooks nodes.
665

666
    """
667
    nl = [self.cfg.GetMasterNode(), self.op.pnode] + self.secondaries
668
    return nl, nl
669

    
670
  def _ReadExportInfo(self):
671
    """Reads the export information from disk.
672

673
    It will override the opcode source node and path with the actual
674
    information, if these two were not specified before.
675

676
    @return: the export information
677

678
    """
679
    assert self.op.mode == constants.INSTANCE_IMPORT
680

    
681
    src_node = self.op.src_node
682
    src_path = self.op.src_path
683

    
684
    if src_node is None:
685
      locked_nodes = self.owned_locks(locking.LEVEL_NODE)
686
      exp_list = self.rpc.call_export_list(locked_nodes)
687
      found = False
688
      for node in exp_list:
689
        if exp_list[node].fail_msg:
690
          continue
691
        if src_path in exp_list[node].payload:
692
          found = True
693
          self.op.src_node = src_node = node
694
          self.op.src_path = src_path = utils.PathJoin(pathutils.EXPORT_DIR,
695
                                                       src_path)
696
          break
697
      if not found:
698
        raise errors.OpPrereqError("No export found for relative path %s" %
699
                                   src_path, errors.ECODE_INVAL)
700

    
701
    CheckNodeOnline(self, src_node)
702
    result = self.rpc.call_export_info(src_node, src_path)
703
    result.Raise("No export or invalid export found in dir %s" % src_path)
704

    
705
    export_info = objects.SerializableConfigParser.Loads(str(result.payload))
706
    if not export_info.has_section(constants.INISECT_EXP):
707
      raise errors.ProgrammerError("Corrupted export config",
708
                                   errors.ECODE_ENVIRON)
709

    
710
    ei_version = export_info.get(constants.INISECT_EXP, "version")
711
    if (int(ei_version) != constants.EXPORT_VERSION):
712
      raise errors.OpPrereqError("Wrong export version %s (wanted %d)" %
713
                                 (ei_version, constants.EXPORT_VERSION),
714
                                 errors.ECODE_ENVIRON)
715
    return export_info
716

    
717
  def _ReadExportParams(self, einfo):
718
    """Use export parameters as defaults.
719

720
    In case the opcode doesn't specify (as in override) some instance
721
    parameters, then try to use them from the export information, if
722
    that declares them.
723

724
    """
725
    self.op.os_type = einfo.get(constants.INISECT_EXP, "os")
726

    
727
    if self.op.disk_template is None:
728
      if einfo.has_option(constants.INISECT_INS, "disk_template"):
729
        self.op.disk_template = einfo.get(constants.INISECT_INS,
730
                                          "disk_template")
731
        if self.op.disk_template not in constants.DISK_TEMPLATES:
732
          raise errors.OpPrereqError("Disk template specified in configuration"
733
                                     " file is not one of the allowed values:"
734
                                     " %s" %
735
                                     " ".join(constants.DISK_TEMPLATES),
736
                                     errors.ECODE_INVAL)
737
      else:
738
        raise errors.OpPrereqError("No disk template specified and the export"
739
                                   " is missing the disk_template information",
740
                                   errors.ECODE_INVAL)
741

    
742
    if not self.op.disks:
743
      disks = []
744
      # TODO: import the disk iv_name too
745
      for idx in range(constants.MAX_DISKS):
746
        if einfo.has_option(constants.INISECT_INS, "disk%d_size" % idx):
747
          disk_sz = einfo.getint(constants.INISECT_INS, "disk%d_size" % idx)
748
          disks.append({constants.IDISK_SIZE: disk_sz})
749
      self.op.disks = disks
750
      if not disks and self.op.disk_template != constants.DT_DISKLESS:
751
        raise errors.OpPrereqError("No disk info specified and the export"
752
                                   " is missing the disk information",
753
                                   errors.ECODE_INVAL)
754

    
755
    if not self.op.nics:
756
      nics = []
757
      for idx in range(constants.MAX_NICS):
758
        if einfo.has_option(constants.INISECT_INS, "nic%d_mac" % idx):
759
          ndict = {}
760
          for name in list(constants.NICS_PARAMETERS) + ["ip", "mac"]:
761
            v = einfo.get(constants.INISECT_INS, "nic%d_%s" % (idx, name))
762
            ndict[name] = v
763
          nics.append(ndict)
764
        else:
765
          break
766
      self.op.nics = nics
767

    
768
    if not self.op.tags and einfo.has_option(constants.INISECT_INS, "tags"):
769
      self.op.tags = einfo.get(constants.INISECT_INS, "tags").split()
770

    
771
    if (self.op.hypervisor is None and
772
        einfo.has_option(constants.INISECT_INS, "hypervisor")):
773
      self.op.hypervisor = einfo.get(constants.INISECT_INS, "hypervisor")
774

    
775
    if einfo.has_section(constants.INISECT_HYP):
776
      # use the export parameters but do not override the ones
777
      # specified by the user
778
      for name, value in einfo.items(constants.INISECT_HYP):
779
        if name not in self.op.hvparams:
780
          self.op.hvparams[name] = value
781

    
782
    if einfo.has_section(constants.INISECT_BEP):
783
      # use the parameters, without overriding
784
      for name, value in einfo.items(constants.INISECT_BEP):
785
        if name not in self.op.beparams:
786
          self.op.beparams[name] = value
787
        # Compatibility for the old "memory" be param
788
        if name == constants.BE_MEMORY:
789
          if constants.BE_MAXMEM not in self.op.beparams:
790
            self.op.beparams[constants.BE_MAXMEM] = value
791
          if constants.BE_MINMEM not in self.op.beparams:
792
            self.op.beparams[constants.BE_MINMEM] = value
793
    else:
794
      # try to read the parameters old style, from the main section
795
      for name in constants.BES_PARAMETERS:
796
        if (name not in self.op.beparams and
797
            einfo.has_option(constants.INISECT_INS, name)):
798
          self.op.beparams[name] = einfo.get(constants.INISECT_INS, name)
799

    
800
    if einfo.has_section(constants.INISECT_OSP):
801
      # use the parameters, without overriding
802
      for name, value in einfo.items(constants.INISECT_OSP):
803
        if name not in self.op.osparams:
804
          self.op.osparams[name] = value
805

    
806
  def _RevertToDefaults(self, cluster):
807
    """Revert the instance parameters to the default values.
808

809
    """
810
    # hvparams
811
    hv_defs = cluster.SimpleFillHV(self.op.hypervisor, self.op.os_type, {})
812
    for name in self.op.hvparams.keys():
813
      if name in hv_defs and hv_defs[name] == self.op.hvparams[name]:
814
        del self.op.hvparams[name]
815
    # beparams
816
    be_defs = cluster.SimpleFillBE({})
817
    for name in self.op.beparams.keys():
818
      if name in be_defs and be_defs[name] == self.op.beparams[name]:
819
        del self.op.beparams[name]
820
    # nic params
821
    nic_defs = cluster.SimpleFillNIC({})
822
    for nic in self.op.nics:
823
      for name in constants.NICS_PARAMETERS:
824
        if name in nic and name in nic_defs and nic[name] == nic_defs[name]:
825
          del nic[name]
826
    # osparams
827
    os_defs = cluster.SimpleFillOS(self.op.os_type, {})
828
    for name in self.op.osparams.keys():
829
      if name in os_defs and os_defs[name] == self.op.osparams[name]:
830
        del self.op.osparams[name]
831

    
832
  def _CalculateFileStorageDir(self):
833
    """Calculate final instance file storage dir.
834

835
    """
836
    # file storage dir calculation/check
837
    self.instance_file_storage_dir = None
838
    if self.op.disk_template in constants.DTS_FILEBASED:
839
      # build the full file storage dir path
840
      joinargs = []
841

    
842
      if self.op.disk_template == constants.DT_SHARED_FILE:
843
        get_fsd_fn = self.cfg.GetSharedFileStorageDir
844
      else:
845
        get_fsd_fn = self.cfg.GetFileStorageDir
846

    
847
      cfg_storagedir = get_fsd_fn()
848
      if not cfg_storagedir:
849
        raise errors.OpPrereqError("Cluster file storage dir not defined",
850
                                   errors.ECODE_STATE)
851
      joinargs.append(cfg_storagedir)
852

    
853
      if self.op.file_storage_dir is not None:
854
        joinargs.append(self.op.file_storage_dir)
855

    
856
      joinargs.append(self.op.instance_name)
857

    
858
      # pylint: disable=W0142
859
      self.instance_file_storage_dir = utils.PathJoin(*joinargs)
860

    
861
  def CheckPrereq(self): # pylint: disable=R0914
862
    """Check prerequisites.
863

864
    """
865
    # Check that the optimistically acquired groups are correct wrt the
866
    # acquired nodes
867
    owned_groups = frozenset(self.owned_locks(locking.LEVEL_NODEGROUP))
868
    owned_nodes = frozenset(self.owned_locks(locking.LEVEL_NODE))
869
    cur_groups = list(self.cfg.GetNodeGroupsFromNodes(owned_nodes))
870
    if not owned_groups.issuperset(cur_groups):
871
      raise errors.OpPrereqError("New instance %s's node groups changed since"
872
                                 " locks were acquired, current groups are"
873
                                 " are '%s', owning groups '%s'; retry the"
874
                                 " operation" %
875
                                 (self.op.instance_name,
876
                                  utils.CommaJoin(cur_groups),
877
                                  utils.CommaJoin(owned_groups)),
878
                                 errors.ECODE_STATE)
879

    
880
    self._CalculateFileStorageDir()
881

    
882
    if self.op.mode == constants.INSTANCE_IMPORT:
883
      export_info = self._ReadExportInfo()
884
      self._ReadExportParams(export_info)
885
      self._old_instance_name = export_info.get(constants.INISECT_INS, "name")
886
    else:
887
      self._old_instance_name = None
888

    
889
    if (not self.cfg.GetVGName() and
890
        self.op.disk_template not in constants.DTS_NOT_LVM):
891
      raise errors.OpPrereqError("Cluster does not support lvm-based"
892
                                 " instances", errors.ECODE_STATE)
893

    
894
    if (self.op.hypervisor is None or
895
        self.op.hypervisor == constants.VALUE_AUTO):
896
      self.op.hypervisor = self.cfg.GetHypervisorType()
897

    
898
    cluster = self.cfg.GetClusterInfo()
899
    enabled_hvs = cluster.enabled_hypervisors
900
    if self.op.hypervisor not in enabled_hvs:
901
      raise errors.OpPrereqError("Selected hypervisor (%s) not enabled in the"
902
                                 " cluster (%s)" %
903
                                 (self.op.hypervisor, ",".join(enabled_hvs)),
904
                                 errors.ECODE_STATE)
905

    
906
    # Check tag validity
907
    for tag in self.op.tags:
908
      objects.TaggableObject.ValidateTag(tag)
909

    
910
    # check hypervisor parameter syntax (locally)
911
    utils.ForceDictType(self.op.hvparams, constants.HVS_PARAMETER_TYPES)
912
    filled_hvp = cluster.SimpleFillHV(self.op.hypervisor, self.op.os_type,
913
                                      self.op.hvparams)
914
    hv_type = hypervisor.GetHypervisorClass(self.op.hypervisor)
915
    hv_type.CheckParameterSyntax(filled_hvp)
916
    self.hv_full = filled_hvp
917
    # check that we don't specify global parameters on an instance
918
    CheckParamsNotGlobal(self.op.hvparams, constants.HVC_GLOBALS, "hypervisor",
919
                         "instance", "cluster")
920

    
921
    # fill and remember the beparams dict
922
    self.be_full = _ComputeFullBeParams(self.op, cluster)
923

    
924
    # build os parameters
925
    self.os_full = cluster.SimpleFillOS(self.op.os_type, self.op.osparams)
926

    
927
    # now that hvp/bep are in final format, let's reset to defaults,
928
    # if told to do so
929
    if self.op.identify_defaults:
930
      self._RevertToDefaults(cluster)
931

    
932
    # NIC buildup
933
    self.nics = _ComputeNics(self.op, cluster, self.check_ip, self.cfg,
934
                             self.proc.GetECId())
935

    
936
    # disk checks/pre-build
937
    default_vg = self.cfg.GetVGName()
938
    self.disks = ComputeDisks(self.op, default_vg)
939

    
940
    if self.op.mode == constants.INSTANCE_IMPORT:
941
      disk_images = []
942
      for idx in range(len(self.disks)):
943
        option = "disk%d_dump" % idx
944
        if export_info.has_option(constants.INISECT_INS, option):
945
          # FIXME: are the old os-es, disk sizes, etc. useful?
946
          export_name = export_info.get(constants.INISECT_INS, option)
947
          image = utils.PathJoin(self.op.src_path, export_name)
948
          disk_images.append(image)
949
        else:
950
          disk_images.append(False)
951

    
952
      self.src_images = disk_images
953

    
954
      if self.op.instance_name == self._old_instance_name:
955
        for idx, nic in enumerate(self.nics):
956
          if nic.mac == constants.VALUE_AUTO:
957
            nic_mac_ini = "nic%d_mac" % idx
958
            nic.mac = export_info.get(constants.INISECT_INS, nic_mac_ini)
959

    
960
    # ENDIF: self.op.mode == constants.INSTANCE_IMPORT
961

    
962
    # ip ping checks (we use the same ip that was resolved in ExpandNames)
963
    if self.op.ip_check:
964
      if netutils.TcpPing(self.check_ip, constants.DEFAULT_NODED_PORT):
965
        raise errors.OpPrereqError("IP %s of instance %s already in use" %
966
                                   (self.check_ip, self.op.instance_name),
967
                                   errors.ECODE_NOTUNIQUE)
968

    
969
    #### mac address generation
970
    # By generating here the mac address both the allocator and the hooks get
971
    # the real final mac address rather than the 'auto' or 'generate' value.
972
    # There is a race condition between the generation and the instance object
973
    # creation, which means that we know the mac is valid now, but we're not
974
    # sure it will be when we actually add the instance. If things go bad
975
    # adding the instance will abort because of a duplicate mac, and the
976
    # creation job will fail.
977
    for nic in self.nics:
978
      if nic.mac in (constants.VALUE_AUTO, constants.VALUE_GENERATE):
979
        nic.mac = self.cfg.GenerateMAC(nic.network, self.proc.GetECId())
980

    
981
    #### allocator run
982

    
983
    if self.op.iallocator is not None:
984
      self._RunAllocator()
985

    
986
    # Release all unneeded node locks
987
    keep_locks = filter(None, [self.op.pnode, self.op.snode, self.op.src_node])
988
    ReleaseLocks(self, locking.LEVEL_NODE, keep=keep_locks)
989
    ReleaseLocks(self, locking.LEVEL_NODE_RES, keep=keep_locks)
990
    ReleaseLocks(self, locking.LEVEL_NODE_ALLOC)
991
    # Release all unneeded group locks
992
    ReleaseLocks(self, locking.LEVEL_NODEGROUP,
993
                 keep=self.cfg.GetNodeGroupsFromNodes(keep_locks))
994

    
995
    assert (self.owned_locks(locking.LEVEL_NODE) ==
996
            self.owned_locks(locking.LEVEL_NODE_RES)), \
997
      "Node locks differ from node resource locks"
998

    
999
    #### node related checks
1000

    
1001
    # check primary node
1002
    self.pnode = pnode = self.cfg.GetNodeInfo(self.op.pnode)
1003
    assert self.pnode is not None, \
1004
      "Cannot retrieve locked node %s" % self.op.pnode
1005
    if pnode.offline:
1006
      raise errors.OpPrereqError("Cannot use offline primary node '%s'" %
1007
                                 pnode.name, errors.ECODE_STATE)
1008
    if pnode.drained:
1009
      raise errors.OpPrereqError("Cannot use drained primary node '%s'" %
1010
                                 pnode.name, errors.ECODE_STATE)
1011
    if not pnode.vm_capable:
1012
      raise errors.OpPrereqError("Cannot use non-vm_capable primary node"
1013
                                 " '%s'" % pnode.name, errors.ECODE_STATE)
1014

    
1015
    self.secondaries = []
1016

    
1017
    # Fill in any IPs from IP pools. This must happen here, because we need to
1018
    # know the nic's primary node, as specified by the iallocator
1019
    for idx, nic in enumerate(self.nics):
1020
      net_uuid = nic.network
1021
      if net_uuid is not None:
1022
        nobj = self.cfg.GetNetwork(net_uuid)
1023
        netparams = self.cfg.GetGroupNetParams(net_uuid, self.pnode.name)
1024
        if netparams is None:
1025
          raise errors.OpPrereqError("No netparams found for network"
1026
                                     " %s. Propably not connected to"
1027
                                     " node's %s nodegroup" %
1028
                                     (nobj.name, self.pnode.name),
1029
                                     errors.ECODE_INVAL)
1030
        self.LogInfo("NIC/%d inherits netparams %s" %
1031
                     (idx, netparams.values()))
1032
        nic.nicparams = dict(netparams)
1033
        if nic.ip is not None:
1034
          if nic.ip.lower() == constants.NIC_IP_POOL:
1035
            try:
1036
              nic.ip = self.cfg.GenerateIp(net_uuid, self.proc.GetECId())
1037
            except errors.ReservationError:
1038
              raise errors.OpPrereqError("Unable to get a free IP for NIC %d"
1039
                                         " from the address pool" % idx,
1040
                                         errors.ECODE_STATE)
1041
            self.LogInfo("Chose IP %s from network %s", nic.ip, nobj.name)
1042
          else:
1043
            try:
1044
              self.cfg.ReserveIp(net_uuid, nic.ip, self.proc.GetECId(),
1045
                                 check=self.op.conflicts_check)
1046
            except errors.ReservationError:
1047
              raise errors.OpPrereqError("IP address %s already in use"
1048
                                         " or does not belong to network %s" %
1049
                                         (nic.ip, nobj.name),
1050
                                         errors.ECODE_NOTUNIQUE)
1051

    
1052
      # net is None, ip None or given
1053
      elif self.op.conflicts_check:
1054
        _CheckForConflictingIp(self, nic.ip, self.pnode.name)
1055

    
1056
    # mirror node verification
1057
    if self.op.disk_template in constants.DTS_INT_MIRROR:
1058
      if self.op.snode == pnode.name:
1059
        raise errors.OpPrereqError("The secondary node cannot be the"
1060
                                   " primary node", errors.ECODE_INVAL)
1061
      CheckNodeOnline(self, self.op.snode)
1062
      CheckNodeNotDrained(self, self.op.snode)
1063
      CheckNodeVmCapable(self, self.op.snode)
1064
      self.secondaries.append(self.op.snode)
1065

    
1066
      snode = self.cfg.GetNodeInfo(self.op.snode)
1067
      if pnode.group != snode.group:
1068
        self.LogWarning("The primary and secondary nodes are in two"
1069
                        " different node groups; the disk parameters"
1070
                        " from the first disk's node group will be"
1071
                        " used")
1072

    
1073
    if not self.op.disk_template in constants.DTS_EXCL_STORAGE:
1074
      nodes = [pnode]
1075
      if self.op.disk_template in constants.DTS_INT_MIRROR:
1076
        nodes.append(snode)
1077
      has_es = lambda n: IsExclusiveStorageEnabledNode(self.cfg, n)
1078
      if compat.any(map(has_es, nodes)):
1079
        raise errors.OpPrereqError("Disk template %s not supported with"
1080
                                   " exclusive storage" % self.op.disk_template,
1081
                                   errors.ECODE_STATE)
1082

    
1083
    nodenames = [pnode.name] + self.secondaries
1084

    
1085
    if not self.adopt_disks:
1086
      if self.op.disk_template == constants.DT_RBD:
1087
        # _CheckRADOSFreeSpace() is just a placeholder.
1088
        # Any function that checks prerequisites can be placed here.
1089
        # Check if there is enough space on the RADOS cluster.
1090
        CheckRADOSFreeSpace()
1091
      elif self.op.disk_template == constants.DT_EXT:
1092
        # FIXME: Function that checks prereqs if needed
1093
        pass
1094
      else:
1095
        # Check lv size requirements, if not adopting
1096
        req_sizes = ComputeDiskSizePerVG(self.op.disk_template, self.disks)
1097
        CheckNodesFreeDiskPerVG(self, nodenames, req_sizes)
1098

    
1099
    elif self.op.disk_template == constants.DT_PLAIN: # Check the adoption data
1100
      all_lvs = set(["%s/%s" % (disk[constants.IDISK_VG],
1101
                                disk[constants.IDISK_ADOPT])
1102
                     for disk in self.disks])
1103
      if len(all_lvs) != len(self.disks):
1104
        raise errors.OpPrereqError("Duplicate volume names given for adoption",
1105
                                   errors.ECODE_INVAL)
1106
      for lv_name in all_lvs:
1107
        try:
1108
          # FIXME: lv_name here is "vg/lv" need to ensure that other calls
1109
          # to ReserveLV uses the same syntax
1110
          self.cfg.ReserveLV(lv_name, self.proc.GetECId())
1111
        except errors.ReservationError:
1112
          raise errors.OpPrereqError("LV named %s used by another instance" %
1113
                                     lv_name, errors.ECODE_NOTUNIQUE)
1114

    
1115
      vg_names = self.rpc.call_vg_list([pnode.name])[pnode.name]
1116
      vg_names.Raise("Cannot get VG information from node %s" % pnode.name)
1117

    
1118
      node_lvs = self.rpc.call_lv_list([pnode.name],
1119
                                       vg_names.payload.keys())[pnode.name]
1120
      node_lvs.Raise("Cannot get LV information from node %s" % pnode.name)
1121
      node_lvs = node_lvs.payload
1122

    
1123
      delta = all_lvs.difference(node_lvs.keys())
1124
      if delta:
1125
        raise errors.OpPrereqError("Missing logical volume(s): %s" %
1126
                                   utils.CommaJoin(delta),
1127
                                   errors.ECODE_INVAL)
1128
      online_lvs = [lv for lv in all_lvs if node_lvs[lv][2]]
1129
      if online_lvs:
1130
        raise errors.OpPrereqError("Online logical volumes found, cannot"
1131
                                   " adopt: %s" % utils.CommaJoin(online_lvs),
1132
                                   errors.ECODE_STATE)
1133
      # update the size of disk based on what is found
1134
      for dsk in self.disks:
1135
        dsk[constants.IDISK_SIZE] = \
1136
          int(float(node_lvs["%s/%s" % (dsk[constants.IDISK_VG],
1137
                                        dsk[constants.IDISK_ADOPT])][0]))
1138

    
1139
    elif self.op.disk_template == constants.DT_BLOCK:
1140
      # Normalize and de-duplicate device paths
1141
      all_disks = set([os.path.abspath(disk[constants.IDISK_ADOPT])
1142
                       for disk in self.disks])
1143
      if len(all_disks) != len(self.disks):
1144
        raise errors.OpPrereqError("Duplicate disk names given for adoption",
1145
                                   errors.ECODE_INVAL)
1146
      baddisks = [d for d in all_disks
1147
                  if not d.startswith(constants.ADOPTABLE_BLOCKDEV_ROOT)]
1148
      if baddisks:
1149
        raise errors.OpPrereqError("Device node(s) %s lie outside %s and"
1150
                                   " cannot be adopted" %
1151
                                   (utils.CommaJoin(baddisks),
1152
                                    constants.ADOPTABLE_BLOCKDEV_ROOT),
1153
                                   errors.ECODE_INVAL)
1154

    
1155
      node_disks = self.rpc.call_bdev_sizes([pnode.name],
1156
                                            list(all_disks))[pnode.name]
1157
      node_disks.Raise("Cannot get block device information from node %s" %
1158
                       pnode.name)
1159
      node_disks = node_disks.payload
1160
      delta = all_disks.difference(node_disks.keys())
1161
      if delta:
1162
        raise errors.OpPrereqError("Missing block device(s): %s" %
1163
                                   utils.CommaJoin(delta),
1164
                                   errors.ECODE_INVAL)
1165
      for dsk in self.disks:
1166
        dsk[constants.IDISK_SIZE] = \
1167
          int(float(node_disks[dsk[constants.IDISK_ADOPT]]))
1168

    
1169
    # Verify instance specs
1170
    spindle_use = self.be_full.get(constants.BE_SPINDLE_USE, None)
1171
    ispec = {
1172
      constants.ISPEC_MEM_SIZE: self.be_full.get(constants.BE_MAXMEM, None),
1173
      constants.ISPEC_CPU_COUNT: self.be_full.get(constants.BE_VCPUS, None),
1174
      constants.ISPEC_DISK_COUNT: len(self.disks),
1175
      constants.ISPEC_DISK_SIZE: [disk[constants.IDISK_SIZE]
1176
                                  for disk in self.disks],
1177
      constants.ISPEC_NIC_COUNT: len(self.nics),
1178
      constants.ISPEC_SPINDLE_USE: spindle_use,
1179
      }
1180

    
1181
    group_info = self.cfg.GetNodeGroup(pnode.group)
1182
    ipolicy = ganeti.masterd.instance.CalculateGroupIPolicy(cluster, group_info)
1183
    res = _ComputeIPolicyInstanceSpecViolation(ipolicy, ispec,
1184
                                               self.op.disk_template)
1185
    if not self.op.ignore_ipolicy and res:
1186
      msg = ("Instance allocation to group %s (%s) violates policy: %s" %
1187
             (pnode.group, group_info.name, utils.CommaJoin(res)))
1188
      raise errors.OpPrereqError(msg, errors.ECODE_INVAL)
1189

    
1190
    CheckHVParams(self, nodenames, self.op.hypervisor, self.op.hvparams)
1191

    
1192
    CheckNodeHasOS(self, pnode.name, self.op.os_type, self.op.force_variant)
1193
    # check OS parameters (remotely)
1194
    CheckOSParams(self, True, nodenames, self.op.os_type, self.os_full)
1195

    
1196
    CheckNicsBridgesExist(self, self.nics, self.pnode.name)
1197

    
1198
    #TODO: _CheckExtParams (remotely)
1199
    # Check parameters for extstorage
1200

    
1201
    # memory check on primary node
1202
    #TODO(dynmem): use MINMEM for checking
1203
    if self.op.start:
1204
      CheckNodeFreeMemory(self, self.pnode.name,
1205
                          "creating instance %s" % self.op.instance_name,
1206
                          self.be_full[constants.BE_MAXMEM],
1207
                          self.op.hypervisor)
1208

    
1209
    self.dry_run_result = list(nodenames)
1210

    
1211
  def Exec(self, feedback_fn):
1212
    """Create and add the instance to the cluster.
1213

1214
    """
1215
    instance = self.op.instance_name
1216
    pnode_name = self.pnode.name
1217

    
1218
    assert not (self.owned_locks(locking.LEVEL_NODE_RES) -
1219
                self.owned_locks(locking.LEVEL_NODE)), \
1220
      "Node locks differ from node resource locks"
1221
    assert not self.glm.is_owned(locking.LEVEL_NODE_ALLOC)
1222

    
1223
    ht_kind = self.op.hypervisor
1224
    if ht_kind in constants.HTS_REQ_PORT:
1225
      network_port = self.cfg.AllocatePort()
1226
    else:
1227
      network_port = None
1228

    
1229
    # This is ugly but we got a chicken-egg problem here
1230
    # We can only take the group disk parameters, as the instance
1231
    # has no disks yet (we are generating them right here).
1232
    node = self.cfg.GetNodeInfo(pnode_name)
1233
    nodegroup = self.cfg.GetNodeGroup(node.group)
1234
    disks = GenerateDiskTemplate(self,
1235
                                 self.op.disk_template,
1236
                                 instance, pnode_name,
1237
                                 self.secondaries,
1238
                                 self.disks,
1239
                                 self.instance_file_storage_dir,
1240
                                 self.op.file_driver,
1241
                                 0,
1242
                                 feedback_fn,
1243
                                 self.cfg.GetGroupDiskParams(nodegroup))
1244

    
1245
    iobj = objects.Instance(name=instance, os=self.op.os_type,
1246
                            primary_node=pnode_name,
1247
                            nics=self.nics, disks=disks,
1248
                            disk_template=self.op.disk_template,
1249
                            disks_active=False,
1250
                            admin_state=constants.ADMINST_DOWN,
1251
                            network_port=network_port,
1252
                            beparams=self.op.beparams,
1253
                            hvparams=self.op.hvparams,
1254
                            hypervisor=self.op.hypervisor,
1255
                            osparams=self.op.osparams,
1256
                            )
1257

    
1258
    if self.op.tags:
1259
      for tag in self.op.tags:
1260
        iobj.AddTag(tag)
1261

    
1262
    if self.adopt_disks:
1263
      if self.op.disk_template == constants.DT_PLAIN:
1264
        # rename LVs to the newly-generated names; we need to construct
1265
        # 'fake' LV disks with the old data, plus the new unique_id
1266
        tmp_disks = [objects.Disk.FromDict(v.ToDict()) for v in disks]
1267
        rename_to = []
1268
        for t_dsk, a_dsk in zip(tmp_disks, self.disks):
1269
          rename_to.append(t_dsk.logical_id)
1270
          t_dsk.logical_id = (t_dsk.logical_id[0], a_dsk[constants.IDISK_ADOPT])
1271
          self.cfg.SetDiskID(t_dsk, pnode_name)
1272
        result = self.rpc.call_blockdev_rename(pnode_name,
1273
                                               zip(tmp_disks, rename_to))
1274
        result.Raise("Failed to rename adoped LVs")
1275
    else:
1276
      feedback_fn("* creating instance disks...")
1277
      try:
1278
        CreateDisks(self, iobj)
1279
      except errors.OpExecError:
1280
        self.LogWarning("Device creation failed")
1281
        self.cfg.ReleaseDRBDMinors(instance)
1282
        raise
1283

    
1284
    feedback_fn("adding instance %s to cluster config" % instance)
1285

    
1286
    self.cfg.AddInstance(iobj, self.proc.GetECId())
1287

    
1288
    # Declare that we don't want to remove the instance lock anymore, as we've
1289
    # added the instance to the config
1290
    del self.remove_locks[locking.LEVEL_INSTANCE]
1291

    
1292
    if self.op.mode == constants.INSTANCE_IMPORT:
1293
      # Release unused nodes
1294
      ReleaseLocks(self, locking.LEVEL_NODE, keep=[self.op.src_node])
1295
    else:
1296
      # Release all nodes
1297
      ReleaseLocks(self, locking.LEVEL_NODE)
1298

    
1299
    disk_abort = False
1300
    if not self.adopt_disks and self.cfg.GetClusterInfo().prealloc_wipe_disks:
1301
      feedback_fn("* wiping instance disks...")
1302
      try:
1303
        WipeDisks(self, iobj)
1304
      except errors.OpExecError, err:
1305
        logging.exception("Wiping disks failed")
1306
        self.LogWarning("Wiping instance disks failed (%s)", err)
1307
        disk_abort = True
1308

    
1309
    if disk_abort:
1310
      # Something is already wrong with the disks, don't do anything else
1311
      pass
1312
    elif self.op.wait_for_sync:
1313
      disk_abort = not WaitForSync(self, iobj)
1314
    elif iobj.disk_template in constants.DTS_INT_MIRROR:
1315
      # make sure the disks are not degraded (still sync-ing is ok)
1316
      feedback_fn("* checking mirrors status")
1317
      disk_abort = not WaitForSync(self, iobj, oneshot=True)
1318
    else:
1319
      disk_abort = False
1320

    
1321
    if disk_abort:
1322
      RemoveDisks(self, iobj)
1323
      self.cfg.RemoveInstance(iobj.name)
1324
      # Make sure the instance lock gets removed
1325
      self.remove_locks[locking.LEVEL_INSTANCE] = iobj.name
1326
      raise errors.OpExecError("There are some degraded disks for"
1327
                               " this instance")
1328

    
1329
    # instance disks are now active
1330
    iobj.disks_active = True
1331

    
1332
    # Release all node resource locks
1333
    ReleaseLocks(self, locking.LEVEL_NODE_RES)
1334

    
1335
    if iobj.disk_template != constants.DT_DISKLESS and not self.adopt_disks:
1336
      # we need to set the disks ID to the primary node, since the
1337
      # preceding code might or might have not done it, depending on
1338
      # disk template and other options
1339
      for disk in iobj.disks:
1340
        self.cfg.SetDiskID(disk, pnode_name)
1341
      if self.op.mode == constants.INSTANCE_CREATE:
1342
        if not self.op.no_install:
1343
          pause_sync = (iobj.disk_template in constants.DTS_INT_MIRROR and
1344
                        not self.op.wait_for_sync)
1345
          if pause_sync:
1346
            feedback_fn("* pausing disk sync to install instance OS")
1347
            result = self.rpc.call_blockdev_pause_resume_sync(pnode_name,
1348
                                                              (iobj.disks,
1349
                                                               iobj), True)
1350
            for idx, success in enumerate(result.payload):
1351
              if not success:
1352
                logging.warn("pause-sync of instance %s for disk %d failed",
1353
                             instance, idx)
1354

    
1355
          feedback_fn("* running the instance OS create scripts...")
1356
          # FIXME: pass debug option from opcode to backend
1357
          os_add_result = \
1358
            self.rpc.call_instance_os_add(pnode_name, (iobj, None), False,
1359
                                          self.op.debug_level)
1360
          if pause_sync:
1361
            feedback_fn("* resuming disk sync")
1362
            result = self.rpc.call_blockdev_pause_resume_sync(pnode_name,
1363
                                                              (iobj.disks,
1364
                                                               iobj), False)
1365
            for idx, success in enumerate(result.payload):
1366
              if not success:
1367
                logging.warn("resume-sync of instance %s for disk %d failed",
1368
                             instance, idx)
1369

    
1370
          os_add_result.Raise("Could not add os for instance %s"
1371
                              " on node %s" % (instance, pnode_name))
1372

    
1373
      else:
1374
        if self.op.mode == constants.INSTANCE_IMPORT:
1375
          feedback_fn("* running the instance OS import scripts...")
1376

    
1377
          transfers = []
1378

    
1379
          for idx, image in enumerate(self.src_images):
1380
            if not image:
1381
              continue
1382

    
1383
            # FIXME: pass debug option from opcode to backend
1384
            dt = masterd.instance.DiskTransfer("disk/%s" % idx,
1385
                                               constants.IEIO_FILE, (image, ),
1386
                                               constants.IEIO_SCRIPT,
1387
                                               (iobj.disks[idx], idx),
1388
                                               None)
1389
            transfers.append(dt)
1390

    
1391
          import_result = \
1392
            masterd.instance.TransferInstanceData(self, feedback_fn,
1393
                                                  self.op.src_node, pnode_name,
1394
                                                  self.pnode.secondary_ip,
1395
                                                  iobj, transfers)
1396
          if not compat.all(import_result):
1397
            self.LogWarning("Some disks for instance %s on node %s were not"
1398
                            " imported successfully" % (instance, pnode_name))
1399

    
1400
          rename_from = self._old_instance_name
1401

    
1402
        elif self.op.mode == constants.INSTANCE_REMOTE_IMPORT:
1403
          feedback_fn("* preparing remote import...")
1404
          # The source cluster will stop the instance before attempting to make
1405
          # a connection. In some cases stopping an instance can take a long
1406
          # time, hence the shutdown timeout is added to the connection
1407
          # timeout.
1408
          connect_timeout = (constants.RIE_CONNECT_TIMEOUT +
1409
                             self.op.source_shutdown_timeout)
1410
          timeouts = masterd.instance.ImportExportTimeouts(connect_timeout)
1411

    
1412
          assert iobj.primary_node == self.pnode.name
1413
          disk_results = \
1414
            masterd.instance.RemoteImport(self, feedback_fn, iobj, self.pnode,
1415
                                          self.source_x509_ca,
1416
                                          self._cds, timeouts)
1417
          if not compat.all(disk_results):
1418
            # TODO: Should the instance still be started, even if some disks
1419
            # failed to import (valid for local imports, too)?
1420
            self.LogWarning("Some disks for instance %s on node %s were not"
1421
                            " imported successfully" % (instance, pnode_name))
1422

    
1423
          rename_from = self.source_instance_name
1424

    
1425
        else:
1426
          # also checked in the prereq part
1427
          raise errors.ProgrammerError("Unknown OS initialization mode '%s'"
1428
                                       % self.op.mode)
1429

    
1430
        # Run rename script on newly imported instance
1431
        assert iobj.name == instance
1432
        feedback_fn("Running rename script for %s" % instance)
1433
        result = self.rpc.call_instance_run_rename(pnode_name, iobj,
1434
                                                   rename_from,
1435
                                                   self.op.debug_level)
1436
        if result.fail_msg:
1437
          self.LogWarning("Failed to run rename script for %s on node"
1438
                          " %s: %s" % (instance, pnode_name, result.fail_msg))
1439

    
1440
    assert not self.owned_locks(locking.LEVEL_NODE_RES)
1441

    
1442
    if self.op.start:
1443
      iobj.admin_state = constants.ADMINST_UP
1444
      self.cfg.Update(iobj, feedback_fn)
1445
      logging.info("Starting instance %s on node %s", instance, pnode_name)
1446
      feedback_fn("* starting instance...")
1447
      result = self.rpc.call_instance_start(pnode_name, (iobj, None, None),
1448
                                            False, self.op.reason)
1449
      result.Raise("Could not start instance")
1450

    
1451
    return list(iobj.all_nodes)
1452

    
1453

    
1454
class LUInstanceRename(LogicalUnit):
1455
  """Rename an instance.
1456

1457
  """
1458
  HPATH = "instance-rename"
1459
  HTYPE = constants.HTYPE_INSTANCE
1460

    
1461
  def CheckArguments(self):
1462
    """Check arguments.
1463

1464
    """
1465
    if self.op.ip_check and not self.op.name_check:
1466
      # TODO: make the ip check more flexible and not depend on the name check
1467
      raise errors.OpPrereqError("IP address check requires a name check",
1468
                                 errors.ECODE_INVAL)
1469

    
1470
  def BuildHooksEnv(self):
1471
    """Build hooks env.
1472

1473
    This runs on master, primary and secondary nodes of the instance.
1474

1475
    """
1476
    env = BuildInstanceHookEnvByObject(self, self.instance)
1477
    env["INSTANCE_NEW_NAME"] = self.op.new_name
1478
    return env
1479

    
1480
  def BuildHooksNodes(self):
1481
    """Build hooks nodes.
1482

1483
    """
1484
    nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
1485
    return (nl, nl)
1486

    
1487
  def CheckPrereq(self):
1488
    """Check prerequisites.
1489

1490
    This checks that the instance is in the cluster and is not running.
1491

1492
    """
1493
    self.op.instance_name = ExpandInstanceName(self.cfg,
1494
                                               self.op.instance_name)
1495
    instance = self.cfg.GetInstanceInfo(self.op.instance_name)
1496
    assert instance is not None
1497
    CheckNodeOnline(self, instance.primary_node)
1498
    CheckInstanceState(self, instance, INSTANCE_NOT_RUNNING,
1499
                       msg="cannot rename")
1500
    self.instance = instance
1501

    
1502
    new_name = self.op.new_name
1503
    if self.op.name_check:
1504
      hostname = _CheckHostnameSane(self, new_name)
1505
      new_name = self.op.new_name = hostname.name
1506
      if (self.op.ip_check and
1507
          netutils.TcpPing(hostname.ip, constants.DEFAULT_NODED_PORT)):
1508
        raise errors.OpPrereqError("IP %s of instance %s already in use" %
1509
                                   (hostname.ip, new_name),
1510
                                   errors.ECODE_NOTUNIQUE)
1511

    
1512
    instance_list = self.cfg.GetInstanceList()
1513
    if new_name in instance_list and new_name != instance.name:
1514
      raise errors.OpPrereqError("Instance '%s' is already in the cluster" %
1515
                                 new_name, errors.ECODE_EXISTS)
1516

    
1517
  def Exec(self, feedback_fn):
1518
    """Rename the instance.
1519

1520
    """
1521
    inst = self.instance
1522
    old_name = inst.name
1523

    
1524
    rename_file_storage = False
1525
    if (inst.disk_template in constants.DTS_FILEBASED and
1526
        self.op.new_name != inst.name):
1527
      old_file_storage_dir = os.path.dirname(inst.disks[0].logical_id[1])
1528
      rename_file_storage = True
1529

    
1530
    self.cfg.RenameInstance(inst.name, self.op.new_name)
1531
    # Change the instance lock. This is definitely safe while we hold the BGL.
1532
    # Otherwise the new lock would have to be added in acquired mode.
1533
    assert self.REQ_BGL
1534
    assert locking.BGL in self.owned_locks(locking.LEVEL_CLUSTER)
1535
    self.glm.remove(locking.LEVEL_INSTANCE, old_name)
1536
    self.glm.add(locking.LEVEL_INSTANCE, self.op.new_name)
1537

    
1538
    # re-read the instance from the configuration after rename
1539
    inst = self.cfg.GetInstanceInfo(self.op.new_name)
1540

    
1541
    if rename_file_storage:
1542
      new_file_storage_dir = os.path.dirname(inst.disks[0].logical_id[1])
1543
      result = self.rpc.call_file_storage_dir_rename(inst.primary_node,
1544
                                                     old_file_storage_dir,
1545
                                                     new_file_storage_dir)
1546
      result.Raise("Could not rename on node %s directory '%s' to '%s'"
1547
                   " (but the instance has been renamed in Ganeti)" %
1548
                   (inst.primary_node, old_file_storage_dir,
1549
                    new_file_storage_dir))
1550

    
1551
    StartInstanceDisks(self, inst, None)
1552
    # update info on disks
1553
    info = GetInstanceInfoText(inst)
1554
    for (idx, disk) in enumerate(inst.disks):
1555
      for node in inst.all_nodes:
1556
        self.cfg.SetDiskID(disk, node)
1557
        result = self.rpc.call_blockdev_setinfo(node, disk, info)
1558
        if result.fail_msg:
1559
          self.LogWarning("Error setting info on node %s for disk %s: %s",
1560
                          node, idx, result.fail_msg)
1561
    try:
1562
      result = self.rpc.call_instance_run_rename(inst.primary_node, inst,
1563
                                                 old_name, self.op.debug_level)
1564
      msg = result.fail_msg
1565
      if msg:
1566
        msg = ("Could not run OS rename script for instance %s on node %s"
1567
               " (but the instance has been renamed in Ganeti): %s" %
1568
               (inst.name, inst.primary_node, msg))
1569
        self.LogWarning(msg)
1570
    finally:
1571
      ShutdownInstanceDisks(self, inst)
1572

    
1573
    return inst.name
1574

    
1575

    
1576
class LUInstanceRemove(LogicalUnit):
1577
  """Remove an instance.
1578

1579
  """
1580
  HPATH = "instance-remove"
1581
  HTYPE = constants.HTYPE_INSTANCE
1582
  REQ_BGL = False
1583

    
1584
  def ExpandNames(self):
1585
    self._ExpandAndLockInstance()
1586
    self.needed_locks[locking.LEVEL_NODE] = []
1587
    self.needed_locks[locking.LEVEL_NODE_RES] = []
1588
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
1589

    
1590
  def DeclareLocks(self, level):
1591
    if level == locking.LEVEL_NODE:
1592
      self._LockInstancesNodes()
1593
    elif level == locking.LEVEL_NODE_RES:
1594
      # Copy node locks
1595
      self.needed_locks[locking.LEVEL_NODE_RES] = \
1596
        CopyLockList(self.needed_locks[locking.LEVEL_NODE])
1597

    
1598
  def BuildHooksEnv(self):
1599
    """Build hooks env.
1600

1601
    This runs on master, primary and secondary nodes of the instance.
1602

1603
    """
1604
    env = BuildInstanceHookEnvByObject(self, self.instance)
1605
    env["SHUTDOWN_TIMEOUT"] = self.op.shutdown_timeout
1606
    return env
1607

    
1608
  def BuildHooksNodes(self):
1609
    """Build hooks nodes.
1610

1611
    """
1612
    nl = [self.cfg.GetMasterNode()]
1613
    nl_post = list(self.instance.all_nodes) + nl
1614
    return (nl, nl_post)
1615

    
1616
  def CheckPrereq(self):
1617
    """Check prerequisites.
1618

1619
    This checks that the instance is in the cluster.
1620

1621
    """
1622
    self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
1623
    assert self.instance is not None, \
1624
      "Cannot retrieve locked instance %s" % self.op.instance_name
1625

    
1626
  def Exec(self, feedback_fn):
1627
    """Remove the instance.
1628

1629
    """
1630
    instance = self.instance
1631
    logging.info("Shutting down instance %s on node %s",
1632
                 instance.name, instance.primary_node)
1633

    
1634
    result = self.rpc.call_instance_shutdown(instance.primary_node, instance,
1635
                                             self.op.shutdown_timeout,
1636
                                             self.op.reason)
1637
    msg = result.fail_msg
1638
    if msg:
1639
      if self.op.ignore_failures:
1640
        feedback_fn("Warning: can't shutdown instance: %s" % msg)
1641
      else:
1642
        raise errors.OpExecError("Could not shutdown instance %s on"
1643
                                 " node %s: %s" %
1644
                                 (instance.name, instance.primary_node, msg))
1645

    
1646
    assert (self.owned_locks(locking.LEVEL_NODE) ==
1647
            self.owned_locks(locking.LEVEL_NODE_RES))
1648
    assert not (set(instance.all_nodes) -
1649
                self.owned_locks(locking.LEVEL_NODE)), \
1650
      "Not owning correct locks"
1651

    
1652
    RemoveInstance(self, feedback_fn, instance, self.op.ignore_failures)
1653

    
1654

    
1655
class LUInstanceMove(LogicalUnit):
1656
  """Move an instance by data-copying.
1657

1658
  """
1659
  HPATH = "instance-move"
1660
  HTYPE = constants.HTYPE_INSTANCE
1661
  REQ_BGL = False
1662

    
1663
  def ExpandNames(self):
1664
    self._ExpandAndLockInstance()
1665
    target_node = ExpandNodeName(self.cfg, self.op.target_node)
1666
    self.op.target_node = target_node
1667
    self.needed_locks[locking.LEVEL_NODE] = [target_node]
1668
    self.needed_locks[locking.LEVEL_NODE_RES] = []
1669
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_APPEND
1670

    
1671
  def DeclareLocks(self, level):
1672
    if level == locking.LEVEL_NODE:
1673
      self._LockInstancesNodes(primary_only=True)
1674
    elif level == locking.LEVEL_NODE_RES:
1675
      # Copy node locks
1676
      self.needed_locks[locking.LEVEL_NODE_RES] = \
1677
        CopyLockList(self.needed_locks[locking.LEVEL_NODE])
1678

    
1679
  def BuildHooksEnv(self):
1680
    """Build hooks env.
1681

1682
    This runs on master, primary and secondary nodes of the instance.
1683

1684
    """
1685
    env = {
1686
      "TARGET_NODE": self.op.target_node,
1687
      "SHUTDOWN_TIMEOUT": self.op.shutdown_timeout,
1688
      }
1689
    env.update(BuildInstanceHookEnvByObject(self, self.instance))
1690
    return env
1691

    
1692
  def BuildHooksNodes(self):
1693
    """Build hooks nodes.
1694

1695
    """
1696
    nl = [
1697
      self.cfg.GetMasterNode(),
1698
      self.instance.primary_node,
1699
      self.op.target_node,
1700
      ]
1701
    return (nl, nl)
1702

    
1703
  def CheckPrereq(self):
1704
    """Check prerequisites.
1705

1706
    This checks that the instance is in the cluster.
1707

1708
    """
1709
    self.instance = instance = self.cfg.GetInstanceInfo(self.op.instance_name)
1710
    assert self.instance is not None, \
1711
      "Cannot retrieve locked instance %s" % self.op.instance_name
1712

    
1713
    if instance.disk_template not in constants.DTS_COPYABLE:
1714
      raise errors.OpPrereqError("Disk template %s not suitable for copying" %
1715
                                 instance.disk_template, errors.ECODE_STATE)
1716

    
1717
    node = self.cfg.GetNodeInfo(self.op.target_node)
1718
    assert node is not None, \
1719
      "Cannot retrieve locked node %s" % self.op.target_node
1720

    
1721
    self.target_node = target_node = node.name
1722

    
1723
    if target_node == instance.primary_node:
1724
      raise errors.OpPrereqError("Instance %s is already on the node %s" %
1725
                                 (instance.name, target_node),
1726
                                 errors.ECODE_STATE)
1727

    
1728
    bep = self.cfg.GetClusterInfo().FillBE(instance)
1729

    
1730
    for idx, dsk in enumerate(instance.disks):
1731
      if dsk.dev_type not in (constants.LD_LV, constants.LD_FILE):
1732
        raise errors.OpPrereqError("Instance disk %d has a complex layout,"
1733
                                   " cannot copy" % idx, errors.ECODE_STATE)
1734

    
1735
    CheckNodeOnline(self, target_node)
1736
    CheckNodeNotDrained(self, target_node)
1737
    CheckNodeVmCapable(self, target_node)
1738
    cluster = self.cfg.GetClusterInfo()
1739
    group_info = self.cfg.GetNodeGroup(node.group)
1740
    ipolicy = ganeti.masterd.instance.CalculateGroupIPolicy(cluster, group_info)
1741
    CheckTargetNodeIPolicy(self, ipolicy, instance, node, self.cfg,
1742
                           ignore=self.op.ignore_ipolicy)
1743

    
1744
    if instance.admin_state == constants.ADMINST_UP:
1745
      # check memory requirements on the secondary node
1746
      CheckNodeFreeMemory(self, target_node,
1747
                          "failing over instance %s" %
1748
                          instance.name, bep[constants.BE_MAXMEM],
1749
                          instance.hypervisor)
1750
    else:
1751
      self.LogInfo("Not checking memory on the secondary node as"
1752
                   " instance will not be started")
1753

    
1754
    # check bridge existance
1755
    CheckInstanceBridgesExist(self, instance, node=target_node)
1756

    
1757
  def Exec(self, feedback_fn):
1758
    """Move an instance.
1759

1760
    The move is done by shutting it down on its present node, copying
1761
    the data over (slow) and starting it on the new node.
1762

1763
    """
1764
    instance = self.instance
1765

    
1766
    source_node = instance.primary_node
1767
    target_node = self.target_node
1768

    
1769
    self.LogInfo("Shutting down instance %s on source node %s",
1770
                 instance.name, source_node)
1771

    
1772
    assert (self.owned_locks(locking.LEVEL_NODE) ==
1773
            self.owned_locks(locking.LEVEL_NODE_RES))
1774

    
1775
    result = self.rpc.call_instance_shutdown(source_node, instance,
1776
                                             self.op.shutdown_timeout,
1777
                                             self.op.reason)
1778
    msg = result.fail_msg
1779
    if msg:
1780
      if self.op.ignore_consistency:
1781
        self.LogWarning("Could not shutdown instance %s on node %s."
1782
                        " Proceeding anyway. Please make sure node"
1783
                        " %s is down. Error details: %s",
1784
                        instance.name, source_node, source_node, msg)
1785
      else:
1786
        raise errors.OpExecError("Could not shutdown instance %s on"
1787
                                 " node %s: %s" %
1788
                                 (instance.name, source_node, msg))
1789

    
1790
    # create the target disks
1791
    try:
1792
      CreateDisks(self, instance, target_node=target_node)
1793
    except errors.OpExecError:
1794
      self.LogWarning("Device creation failed")
1795
      self.cfg.ReleaseDRBDMinors(instance.name)
1796
      raise
1797

    
1798
    cluster_name = self.cfg.GetClusterInfo().cluster_name
1799

    
1800
    errs = []
1801
    # activate, get path, copy the data over
1802
    for idx, disk in enumerate(instance.disks):
1803
      self.LogInfo("Copying data for disk %d", idx)
1804
      result = self.rpc.call_blockdev_assemble(target_node, (disk, instance),
1805
                                               instance.name, True, idx)
1806
      if result.fail_msg:
1807
        self.LogWarning("Can't assemble newly created disk %d: %s",
1808
                        idx, result.fail_msg)
1809
        errs.append(result.fail_msg)
1810
        break
1811
      dev_path, _ = result.payload
1812
      result = self.rpc.call_blockdev_export(source_node, (disk, instance),
1813
                                             target_node, dev_path,
1814
                                             cluster_name)
1815
      if result.fail_msg:
1816
        self.LogWarning("Can't copy data over for disk %d: %s",
1817
                        idx, result.fail_msg)
1818
        errs.append(result.fail_msg)
1819
        break
1820

    
1821
    if errs:
1822
      self.LogWarning("Some disks failed to copy, aborting")
1823
      try:
1824
        RemoveDisks(self, instance, target_node=target_node)
1825
      finally:
1826
        self.cfg.ReleaseDRBDMinors(instance.name)
1827
        raise errors.OpExecError("Errors during disk copy: %s" %
1828
                                 (",".join(errs),))
1829

    
1830
    instance.primary_node = target_node
1831
    self.cfg.Update(instance, feedback_fn)
1832

    
1833
    self.LogInfo("Removing the disks on the original node")
1834
    RemoveDisks(self, instance, target_node=source_node)
1835

    
1836
    # Only start the instance if it's marked as up
1837
    if instance.admin_state == constants.ADMINST_UP:
1838
      self.LogInfo("Starting instance %s on node %s",
1839
                   instance.name, target_node)
1840

    
1841
      disks_ok, _ = AssembleInstanceDisks(self, instance,
1842
                                          ignore_secondaries=True)
1843
      if not disks_ok:
1844
        ShutdownInstanceDisks(self, instance)
1845
        raise errors.OpExecError("Can't activate the instance's disks")
1846

    
1847
      result = self.rpc.call_instance_start(target_node,
1848
                                            (instance, None, None), False,
1849
                                            self.op.reason)
1850
      msg = result.fail_msg
1851
      if msg:
1852
        ShutdownInstanceDisks(self, instance)
1853
        raise errors.OpExecError("Could not start instance %s on node %s: %s" %
1854
                                 (instance.name, target_node, msg))
1855

    
1856

    
1857
class LUInstanceMultiAlloc(NoHooksLU):
1858
  """Allocates multiple instances at the same time.
1859

1860
  """
1861
  REQ_BGL = False
1862

    
1863
  def CheckArguments(self):
1864
    """Check arguments.
1865

1866
    """
1867
    nodes = []
1868
    for inst in self.op.instances:
1869
      if inst.iallocator is not None:
1870
        raise errors.OpPrereqError("iallocator are not allowed to be set on"
1871
                                   " instance objects", errors.ECODE_INVAL)
1872
      nodes.append(bool(inst.pnode))
1873
      if inst.disk_template in constants.DTS_INT_MIRROR:
1874
        nodes.append(bool(inst.snode))
1875

    
1876
    has_nodes = compat.any(nodes)
1877
    if compat.all(nodes) ^ has_nodes:
1878
      raise errors.OpPrereqError("There are instance objects providing"
1879
                                 " pnode/snode while others do not",
1880
                                 errors.ECODE_INVAL)
1881

    
1882
    if not has_nodes and self.op.iallocator is None:
1883
      default_iallocator = self.cfg.GetDefaultIAllocator()
1884
      if default_iallocator:
1885
        self.op.iallocator = default_iallocator
1886
      else:
1887
        raise errors.OpPrereqError("No iallocator or nodes on the instances"
1888
                                   " given and no cluster-wide default"
1889
                                   " iallocator found; please specify either"
1890
                                   " an iallocator or nodes on the instances"
1891
                                   " or set a cluster-wide default iallocator",
1892
                                   errors.ECODE_INVAL)
1893

    
1894
    _CheckOpportunisticLocking(self.op)
1895

    
1896
    dups = utils.FindDuplicates([op.instance_name for op in self.op.instances])
1897
    if dups:
1898
      raise errors.OpPrereqError("There are duplicate instance names: %s" %
1899
                                 utils.CommaJoin(dups), errors.ECODE_INVAL)
1900

    
1901
  def ExpandNames(self):
1902
    """Calculate the locks.
1903

1904
    """
1905
    self.share_locks = ShareAll()
1906
    self.needed_locks = {
1907
      # iallocator will select nodes and even if no iallocator is used,
1908
      # collisions with LUInstanceCreate should be avoided
1909
      locking.LEVEL_NODE_ALLOC: locking.ALL_SET,
1910
      }
1911

    
1912
    if self.op.iallocator:
1913
      self.needed_locks[locking.LEVEL_NODE] = locking.ALL_SET
1914
      self.needed_locks[locking.LEVEL_NODE_RES] = locking.ALL_SET
1915

    
1916
      if self.op.opportunistic_locking:
1917
        self.opportunistic_locks[locking.LEVEL_NODE] = True
1918
    else:
1919
      nodeslist = []
1920
      for inst in self.op.instances:
1921
        inst.pnode = ExpandNodeName(self.cfg, inst.pnode)
1922
        nodeslist.append(inst.pnode)
1923
        if inst.snode is not None:
1924
          inst.snode = ExpandNodeName(self.cfg, inst.snode)
1925
          nodeslist.append(inst.snode)
1926

    
1927
      self.needed_locks[locking.LEVEL_NODE] = nodeslist
1928
      # Lock resources of instance's primary and secondary nodes (copy to
1929
      # prevent accidential modification)
1930
      self.needed_locks[locking.LEVEL_NODE_RES] = list(nodeslist)
1931

    
1932
  def DeclareLocks(self, level):
1933
    if level == locking.LEVEL_NODE_RES and \
1934
      self.opportunistic_locks[locking.LEVEL_NODE]:
1935
      # Even when using opportunistic locking, we require the same set of
1936
      # NODE_RES locks as we got NODE locks
1937
      self.needed_locks[locking.LEVEL_NODE_RES] = \
1938
        self.owned_locks(locking.LEVEL_NODE)
1939

    
1940
  def CheckPrereq(self):
1941
    """Check prerequisite.
1942

1943
    """
1944
    if self.op.iallocator:
1945
      cluster = self.cfg.GetClusterInfo()
1946
      default_vg = self.cfg.GetVGName()
1947
      ec_id = self.proc.GetECId()
1948

    
1949
      if self.op.opportunistic_locking:
1950
        # Only consider nodes for which a lock is held
1951
        node_whitelist = list(self.owned_locks(locking.LEVEL_NODE))
1952
      else:
1953
        node_whitelist = None
1954

    
1955
      insts = [_CreateInstanceAllocRequest(op, ComputeDisks(op, default_vg),
1956
                                           _ComputeNics(op, cluster, None,
1957
                                                        self.cfg, ec_id),
1958
                                           _ComputeFullBeParams(op, cluster),
1959
                                           node_whitelist)
1960
               for op in self.op.instances]
1961

    
1962
      req = iallocator.IAReqMultiInstanceAlloc(instances=insts)
1963
      ial = iallocator.IAllocator(self.cfg, self.rpc, req)
1964

    
1965
      ial.Run(self.op.iallocator)
1966

    
1967
      if not ial.success:
1968
        raise errors.OpPrereqError("Can't compute nodes using"
1969
                                   " iallocator '%s': %s" %
1970
                                   (self.op.iallocator, ial.info),
1971
                                   errors.ECODE_NORES)
1972

    
1973
      self.ia_result = ial.result
1974

    
1975
    if self.op.dry_run:
1976
      self.dry_run_result = objects.FillDict(self._ConstructPartialResult(), {
1977
        constants.JOB_IDS_KEY: [],
1978
        })
1979

    
1980
  def _ConstructPartialResult(self):
1981
    """Contructs the partial result.
1982

1983
    """
1984
    if self.op.iallocator:
1985
      (allocatable, failed_insts) = self.ia_result
1986
      allocatable_insts = map(compat.fst, allocatable)
1987
    else:
1988
      allocatable_insts = [op.instance_name for op in self.op.instances]
1989
      failed_insts = []
1990

    
1991
    return {
1992
      opcodes.OpInstanceMultiAlloc.ALLOCATABLE_KEY: allocatable_insts,
1993
      opcodes.OpInstanceMultiAlloc.FAILED_KEY: failed_insts,
1994
      }
1995

    
1996
  def Exec(self, feedback_fn):
1997
    """Executes the opcode.
1998

1999
    """
2000
    jobs = []
2001
    if self.op.iallocator:
2002
      op2inst = dict((op.instance_name, op) for op in self.op.instances)
2003
      (allocatable, failed) = self.ia_result
2004

    
2005
      for (name, nodes) in allocatable:
2006
        op = op2inst.pop(name)
2007

    
2008
        if len(nodes) > 1:
2009
          (op.pnode, op.snode) = nodes
2010
        else:
2011
          (op.pnode,) = nodes
2012

    
2013
        jobs.append([op])
2014

    
2015
      missing = set(op2inst.keys()) - set(failed)
2016
      assert not missing, \
2017
        "Iallocator did return incomplete result: %s" % \
2018
        utils.CommaJoin(missing)
2019
    else:
2020
      jobs.extend([op] for op in self.op.instances)
2021

    
2022
    return ResultWithJobs(jobs, **self._ConstructPartialResult())
2023

    
2024

    
2025
class _InstNicModPrivate:
2026
  """Data structure for network interface modifications.
2027

2028
  Used by L{LUInstanceSetParams}.
2029

2030
  """
2031
  def __init__(self):
2032
    self.params = None
2033
    self.filled = None
2034

    
2035

    
2036
def _PrepareContainerMods(mods, private_fn):
2037
  """Prepares a list of container modifications by adding a private data field.
2038

2039
  @type mods: list of tuples; (operation, index, parameters)
2040
  @param mods: List of modifications
2041
  @type private_fn: callable or None
2042
  @param private_fn: Callable for constructing a private data field for a
2043
    modification
2044
  @rtype: list
2045

2046
  """
2047
  if private_fn is None:
2048
    fn = lambda: None
2049
  else:
2050
    fn = private_fn
2051

    
2052
  return [(op, idx, params, fn()) for (op, idx, params) in mods]
2053

    
2054

    
2055
def _CheckNodesPhysicalCPUs(lu, nodenames, requested, hypervisor_name):
2056
  """Checks if nodes have enough physical CPUs
2057

2058
  This function checks if all given nodes have the needed number of
2059
  physical CPUs. In case any node has less CPUs or we cannot get the
2060
  information from the node, this function raises an OpPrereqError
2061
  exception.
2062

2063
  @type lu: C{LogicalUnit}
2064
  @param lu: a logical unit from which we get configuration data
2065
  @type nodenames: C{list}
2066
  @param nodenames: the list of node names to check
2067
  @type requested: C{int}
2068
  @param requested: the minimum acceptable number of physical CPUs
2069
  @raise errors.OpPrereqError: if the node doesn't have enough CPUs,
2070
      or we cannot check the node
2071

2072
  """
2073
  nodeinfo = lu.rpc.call_node_info(nodenames, None, [hypervisor_name], None)
2074
  for node in nodenames:
2075
    info = nodeinfo[node]
2076
    info.Raise("Cannot get current information from node %s" % node,
2077
               prereq=True, ecode=errors.ECODE_ENVIRON)
2078
    (_, _, (hv_info, )) = info.payload
2079
    num_cpus = hv_info.get("cpu_total", None)
2080
    if not isinstance(num_cpus, int):
2081
      raise errors.OpPrereqError("Can't compute the number of physical CPUs"
2082
                                 " on node %s, result was '%s'" %
2083
                                 (node, num_cpus), errors.ECODE_ENVIRON)
2084
    if requested > num_cpus:
2085
      raise errors.OpPrereqError("Node %s has %s physical CPUs, but %s are "
2086
                                 "required" % (node, num_cpus, requested),
2087
                                 errors.ECODE_NORES)
2088

    
2089

    
2090
def GetItemFromContainer(identifier, kind, container):
2091
  """Return the item refered by the identifier.
2092

2093
  @type identifier: string
2094
  @param identifier: Item index or name or UUID
2095
  @type kind: string
2096
  @param kind: One-word item description
2097
  @type container: list
2098
  @param container: Container to get the item from
2099

2100
  """
2101
  # Index
2102
  try:
2103
    idx = int(identifier)
2104
    if idx == -1:
2105
      # Append
2106
      absidx = len(container) - 1
2107
    elif idx < 0:
2108
      raise IndexError("Not accepting negative indices other than -1")
2109
    elif idx > len(container):
2110
      raise IndexError("Got %s index %s, but there are only %s" %
2111
                       (kind, idx, len(container)))
2112
    else:
2113
      absidx = idx
2114
    return (absidx, container[idx])
2115
  except ValueError:
2116
    pass
2117

    
2118
  for idx, item in enumerate(container):
2119
    if item.uuid == identifier or item.name == identifier:
2120
      return (idx, item)
2121

    
2122
  raise errors.OpPrereqError("Cannot find %s with identifier %s" %
2123
                             (kind, identifier), errors.ECODE_NOENT)
2124

    
2125

    
2126
def _ApplyContainerMods(kind, container, chgdesc, mods,
2127
                        create_fn, modify_fn, remove_fn):
2128
  """Applies descriptions in C{mods} to C{container}.
2129

2130
  @type kind: string
2131
  @param kind: One-word item description
2132
  @type container: list
2133
  @param container: Container to modify
2134
  @type chgdesc: None or list
2135
  @param chgdesc: List of applied changes
2136
  @type mods: list
2137
  @param mods: Modifications as returned by L{_PrepareContainerMods}
2138
  @type create_fn: callable
2139
  @param create_fn: Callback for creating a new item (L{constants.DDM_ADD});
2140
    receives absolute item index, parameters and private data object as added
2141
    by L{_PrepareContainerMods}, returns tuple containing new item and changes
2142
    as list
2143
  @type modify_fn: callable
2144
  @param modify_fn: Callback for modifying an existing item
2145
    (L{constants.DDM_MODIFY}); receives absolute item index, item, parameters
2146
    and private data object as added by L{_PrepareContainerMods}, returns
2147
    changes as list
2148
  @type remove_fn: callable
2149
  @param remove_fn: Callback on removing item; receives absolute item index,
2150
    item and private data object as added by L{_PrepareContainerMods}
2151

2152
  """
2153
  for (op, identifier, params, private) in mods:
2154
    changes = None
2155

    
2156
    if op == constants.DDM_ADD:
2157
      # Calculate where item will be added
2158
      # When adding an item, identifier can only be an index
2159
      try:
2160
        idx = int(identifier)
2161
      except ValueError:
2162
        raise errors.OpPrereqError("Only possitive integer or -1 is accepted as"
2163
                                   " identifier for %s" % constants.DDM_ADD,
2164
                                   errors.ECODE_INVAL)
2165
      if idx == -1:
2166
        addidx = len(container)
2167
      else:
2168
        if idx < 0:
2169
          raise IndexError("Not accepting negative indices other than -1")
2170
        elif idx > len(container):
2171
          raise IndexError("Got %s index %s, but there are only %s" %
2172
                           (kind, idx, len(container)))
2173
        addidx = idx
2174

    
2175
      if create_fn is None:
2176
        item = params
2177
      else:
2178
        (item, changes) = create_fn(addidx, params, private)
2179

    
2180
      if idx == -1:
2181
        container.append(item)
2182
      else:
2183
        assert idx >= 0
2184
        assert idx <= len(container)
2185
        # list.insert does so before the specified index
2186
        container.insert(idx, item)
2187
    else:
2188
      # Retrieve existing item
2189
      (absidx, item) = GetItemFromContainer(identifier, kind, container)
2190

    
2191
      if op == constants.DDM_REMOVE:
2192
        assert not params
2193

    
2194
        if remove_fn is not None:
2195
          remove_fn(absidx, item, private)
2196

    
2197
        changes = [("%s/%s" % (kind, absidx), "remove")]
2198

    
2199
        assert container[absidx] == item
2200
        del container[absidx]
2201
      elif op == constants.DDM_MODIFY:
2202
        if modify_fn is not None:
2203
          changes = modify_fn(absidx, item, params, private)
2204
      else:
2205
        raise errors.ProgrammerError("Unhandled operation '%s'" % op)
2206

    
2207
    assert _TApplyContModsCbChanges(changes)
2208

    
2209
    if not (chgdesc is None or changes is None):
2210
      chgdesc.extend(changes)
2211

    
2212

    
2213
def _UpdateIvNames(base_index, disks):
2214
  """Updates the C{iv_name} attribute of disks.
2215

2216
  @type disks: list of L{objects.Disk}
2217

2218
  """
2219
  for (idx, disk) in enumerate(disks):
2220
    disk.iv_name = "disk/%s" % (base_index + idx, )
2221

    
2222

    
2223
class LUInstanceSetParams(LogicalUnit):
2224
  """Modifies an instances's parameters.
2225

2226
  """
2227
  HPATH = "instance-modify"
2228
  HTYPE = constants.HTYPE_INSTANCE
2229
  REQ_BGL = False
2230

    
2231
  @staticmethod
2232
  def _UpgradeDiskNicMods(kind, mods, verify_fn):
2233
    assert ht.TList(mods)
2234
    assert not mods or len(mods[0]) in (2, 3)
2235

    
2236
    if mods and len(mods[0]) == 2:
2237
      result = []
2238

    
2239
      addremove = 0
2240
      for op, params in mods:
2241
        if op in (constants.DDM_ADD, constants.DDM_REMOVE):
2242
          result.append((op, -1, params))
2243
          addremove += 1
2244

    
2245
          if addremove > 1:
2246
            raise errors.OpPrereqError("Only one %s add or remove operation is"
2247
                                       " supported at a time" % kind,
2248
                                       errors.ECODE_INVAL)
2249
        else:
2250
          result.append((constants.DDM_MODIFY, op, params))
2251

    
2252
      assert verify_fn(result)
2253
    else:
2254
      result = mods
2255

    
2256
    return result
2257

    
2258
  @staticmethod
2259
  def _CheckMods(kind, mods, key_types, item_fn):
2260
    """Ensures requested disk/NIC modifications are valid.
2261

2262
    """
2263
    for (op, _, params) in mods:
2264
      assert ht.TDict(params)
2265

    
2266
      # If 'key_types' is an empty dict, we assume we have an
2267
      # 'ext' template and thus do not ForceDictType
2268
      if key_types:
2269
        utils.ForceDictType(params, key_types)
2270

    
2271
      if op == constants.DDM_REMOVE:
2272
        if params:
2273
          raise errors.OpPrereqError("No settings should be passed when"
2274
                                     " removing a %s" % kind,
2275
                                     errors.ECODE_INVAL)
2276
      elif op in (constants.DDM_ADD, constants.DDM_MODIFY):
2277
        item_fn(op, params)
2278
      else:
2279
        raise errors.ProgrammerError("Unhandled operation '%s'" % op)
2280

    
2281
  @staticmethod
2282
  def _VerifyDiskModification(op, params):
2283
    """Verifies a disk modification.
2284

2285
    """
2286
    if op == constants.DDM_ADD:
2287
      mode = params.setdefault(constants.IDISK_MODE, constants.DISK_RDWR)
2288
      if mode not in constants.DISK_ACCESS_SET:
2289
        raise errors.OpPrereqError("Invalid disk access mode '%s'" % mode,
2290
                                   errors.ECODE_INVAL)
2291

    
2292
      size = params.get(constants.IDISK_SIZE, None)
2293
      if size is None:
2294
        raise errors.OpPrereqError("Required disk parameter '%s' missing" %
2295
                                   constants.IDISK_SIZE, errors.ECODE_INVAL)
2296

    
2297
      try:
2298
        size = int(size)
2299
      except (TypeError, ValueError), err:
2300
        raise errors.OpPrereqError("Invalid disk size parameter: %s" % err,
2301
                                   errors.ECODE_INVAL)
2302

    
2303
      params[constants.IDISK_SIZE] = size
2304
      name = params.get(constants.IDISK_NAME, None)
2305
      if name is not None and name.lower() == constants.VALUE_NONE:
2306
        params[constants.IDISK_NAME] = None
2307

    
2308
    elif op == constants.DDM_MODIFY:
2309
      if constants.IDISK_SIZE in params:
2310
        raise errors.OpPrereqError("Disk size change not possible, use"
2311
                                   " grow-disk", errors.ECODE_INVAL)
2312
      name = params.get(constants.IDISK_NAME, None)
2313
      if name is not None and name.lower() == constants.VALUE_NONE:
2314
        params[constants.IDISK_NAME] = None
2315

    
2316
  @staticmethod
2317
  def _VerifyNicModification(op, params):
2318
    """Verifies a network interface modification.
2319

2320
    """
2321
    if op in (constants.DDM_ADD, constants.DDM_MODIFY):
2322
      ip = params.get(constants.INIC_IP, None)
2323
      name = params.get(constants.INIC_NAME, None)
2324
      req_net = params.get(constants.INIC_NETWORK, None)
2325
      link = params.get(constants.NIC_LINK, None)
2326
      mode = params.get(constants.NIC_MODE, None)
2327
      if name is not None and name.lower() == constants.VALUE_NONE:
2328
        params[constants.INIC_NAME] = None
2329
      if req_net is not None:
2330
        if req_net.lower() == constants.VALUE_NONE:
2331
          params[constants.INIC_NETWORK] = None
2332
          req_net = None
2333
        elif link is not None or mode is not None:
2334
          raise errors.OpPrereqError("If network is given"
2335
                                     " mode or link should not",
2336
                                     errors.ECODE_INVAL)
2337

    
2338
      if op == constants.DDM_ADD:
2339
        macaddr = params.get(constants.INIC_MAC, None)
2340
        if macaddr is None:
2341
          params[constants.INIC_MAC] = constants.VALUE_AUTO
2342

    
2343
      if ip is not None:
2344
        if ip.lower() == constants.VALUE_NONE:
2345
          params[constants.INIC_IP] = None
2346
        else:
2347
          if ip.lower() == constants.NIC_IP_POOL:
2348
            if op == constants.DDM_ADD and req_net is None:
2349
              raise errors.OpPrereqError("If ip=pool, parameter network"
2350
                                         " cannot be none",
2351
                                         errors.ECODE_INVAL)
2352
          else:
2353
            if not netutils.IPAddress.IsValid(ip):
2354
              raise errors.OpPrereqError("Invalid IP address '%s'" % ip,
2355
                                         errors.ECODE_INVAL)
2356

    
2357
      if constants.INIC_MAC in params:
2358
        macaddr = params[constants.INIC_MAC]
2359
        if macaddr not in (constants.VALUE_AUTO, constants.VALUE_GENERATE):
2360
          macaddr = utils.NormalizeAndValidateMac(macaddr)
2361

    
2362
        if op == constants.DDM_MODIFY and macaddr == constants.VALUE_AUTO:
2363
          raise errors.OpPrereqError("'auto' is not a valid MAC address when"
2364
                                     " modifying an existing NIC",
2365
                                     errors.ECODE_INVAL)
2366

    
2367
  def CheckArguments(self):
2368
    if not (self.op.nics or self.op.disks or self.op.disk_template or
2369
            self.op.hvparams or self.op.beparams or self.op.os_name or
2370
            self.op.osparams or self.op.offline is not None or
2371
            self.op.runtime_mem or self.op.pnode):
2372
      raise errors.OpPrereqError("No changes submitted", errors.ECODE_INVAL)
2373

    
2374
    if self.op.hvparams:
2375
      CheckParamsNotGlobal(self.op.hvparams, constants.HVC_GLOBALS,
2376
                           "hypervisor", "instance", "cluster")
2377

    
2378
    self.op.disks = self._UpgradeDiskNicMods(
2379
      "disk", self.op.disks, opcodes.OpInstanceSetParams.TestDiskModifications)
2380
    self.op.nics = self._UpgradeDiskNicMods(
2381
      "NIC", self.op.nics, opcodes.OpInstanceSetParams.TestNicModifications)
2382

    
2383
    if self.op.disks and self.op.disk_template is not None:
2384
      raise errors.OpPrereqError("Disk template conversion and other disk"
2385
                                 " changes not supported at the same time",
2386
                                 errors.ECODE_INVAL)
2387

    
2388
    if (self.op.disk_template and
2389
        self.op.disk_template in constants.DTS_INT_MIRROR and
2390
        self.op.remote_node is None):
2391
      raise errors.OpPrereqError("Changing the disk template to a mirrored"
2392
                                 " one requires specifying a secondary node",
2393
                                 errors.ECODE_INVAL)
2394

    
2395
    # Check NIC modifications
2396
    self._CheckMods("NIC", self.op.nics, constants.INIC_PARAMS_TYPES,
2397
                    self._VerifyNicModification)
2398

    
2399
    if self.op.pnode:
2400
      self.op.pnode = ExpandNodeName(self.cfg, self.op.pnode)
2401

    
2402
  def ExpandNames(self):
2403
    self._ExpandAndLockInstance()
2404
    self.needed_locks[locking.LEVEL_NODEGROUP] = []
2405
    # Can't even acquire node locks in shared mode as upcoming changes in
2406
    # Ganeti 2.6 will start to modify the node object on disk conversion
2407
    self.needed_locks[locking.LEVEL_NODE] = []
2408
    self.needed_locks[locking.LEVEL_NODE_RES] = []
2409
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
2410
    # Look node group to look up the ipolicy
2411
    self.share_locks[locking.LEVEL_NODEGROUP] = 1
2412

    
2413
  def DeclareLocks(self, level):
2414
    if level == locking.LEVEL_NODEGROUP:
2415
      assert not self.needed_locks[locking.LEVEL_NODEGROUP]
2416
      # Acquire locks for the instance's nodegroups optimistically. Needs
2417
      # to be verified in CheckPrereq
2418
      self.needed_locks[locking.LEVEL_NODEGROUP] = \
2419
        self.cfg.GetInstanceNodeGroups(self.op.instance_name)
2420
    elif level == locking.LEVEL_NODE:
2421
      self._LockInstancesNodes()
2422
      if self.op.disk_template and self.op.remote_node:
2423
        self.op.remote_node = ExpandNodeName(self.cfg, self.op.remote_node)
2424
        self.needed_locks[locking.LEVEL_NODE].append(self.op.remote_node)
2425
    elif level == locking.LEVEL_NODE_RES and self.op.disk_template:
2426
      # Copy node locks
2427
      self.needed_locks[locking.LEVEL_NODE_RES] = \
2428
        CopyLockList(self.needed_locks[locking.LEVEL_NODE])
2429

    
2430
  def BuildHooksEnv(self):
2431
    """Build hooks env.
2432

2433
    This runs on the master, primary and secondaries.
2434

2435
    """
2436
    args = {}
2437
    if constants.BE_MINMEM in self.be_new:
2438
      args["minmem"] = self.be_new[constants.BE_MINMEM]
2439
    if constants.BE_MAXMEM in self.be_new:
2440
      args["maxmem"] = self.be_new[constants.BE_MAXMEM]
2441
    if constants.BE_VCPUS in self.be_new:
2442
      args["vcpus"] = self.be_new[constants.BE_VCPUS]
2443
    # TODO: export disk changes. Note: _BuildInstanceHookEnv* don't export disk
2444
    # information at all.
2445

    
2446
    if self._new_nics is not None:
2447
      nics = []
2448

    
2449
      for nic in self._new_nics:
2450
        n = copy.deepcopy(nic)
2451
        nicparams = self.cluster.SimpleFillNIC(n.nicparams)
2452
        n.nicparams = nicparams
2453
        nics.append(NICToTuple(self, n))
2454

    
2455
      args["nics"] = nics
2456

    
2457
    env = BuildInstanceHookEnvByObject(self, self.instance, override=args)
2458
    if self.op.disk_template:
2459
      env["NEW_DISK_TEMPLATE"] = self.op.disk_template
2460
    if self.op.runtime_mem:
2461
      env["RUNTIME_MEMORY"] = self.op.runtime_mem
2462

    
2463
    return env
2464

    
2465
  def BuildHooksNodes(self):
2466
    """Build hooks nodes.
2467

2468
    """
2469
    nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
2470
    return (nl, nl)
2471

    
2472
  def _PrepareNicModification(self, params, private, old_ip, old_net_uuid,
2473
                              old_params, cluster, pnode):
2474

    
2475
    update_params_dict = dict([(key, params[key])
2476
                               for key in constants.NICS_PARAMETERS
2477
                               if key in params])
2478

    
2479
    req_link = update_params_dict.get(constants.NIC_LINK, None)
2480
    req_mode = update_params_dict.get(constants.NIC_MODE, None)
2481

    
2482
    new_net_uuid = None
2483
    new_net_uuid_or_name = params.get(constants.INIC_NETWORK, old_net_uuid)
2484
    if new_net_uuid_or_name:
2485
      new_net_uuid = self.cfg.LookupNetwork(new_net_uuid_or_name)
2486
      new_net_obj = self.cfg.GetNetwork(new_net_uuid)
2487

    
2488
    if old_net_uuid:
2489
      old_net_obj = self.cfg.GetNetwork(old_net_uuid)
2490

    
2491
    if new_net_uuid:
2492
      netparams = self.cfg.GetGroupNetParams(new_net_uuid, pnode)
2493
      if not netparams:
2494
        raise errors.OpPrereqError("No netparams found for the network"
2495
                                   " %s, probably not connected" %
2496
                                   new_net_obj.name, errors.ECODE_INVAL)
2497
      new_params = dict(netparams)
2498
    else:
2499
      new_params = GetUpdatedParams(old_params, update_params_dict)
2500

    
2501
    utils.ForceDictType(new_params, constants.NICS_PARAMETER_TYPES)
2502

    
2503
    new_filled_params = cluster.SimpleFillNIC(new_params)
2504
    objects.NIC.CheckParameterSyntax(new_filled_params)
2505

    
2506
    new_mode = new_filled_params[constants.NIC_MODE]
2507
    if new_mode == constants.NIC_MODE_BRIDGED:
2508
      bridge = new_filled_params[constants.NIC_LINK]
2509
      msg = self.rpc.call_bridges_exist(pnode, [bridge]).fail_msg
2510
      if msg:
2511
        msg = "Error checking bridges on node '%s': %s" % (pnode, msg)
2512
        if self.op.force:
2513
          self.warn.append(msg)
2514
        else:
2515
          raise errors.OpPrereqError(msg, errors.ECODE_ENVIRON)
2516

    
2517
    elif new_mode == constants.NIC_MODE_ROUTED:
2518
      ip = params.get(constants.INIC_IP, old_ip)
2519
      if ip is None:
2520
        raise errors.OpPrereqError("Cannot set the NIC IP address to None"
2521
                                   " on a routed NIC", errors.ECODE_INVAL)
2522

    
2523
    elif new_mode == constants.NIC_MODE_OVS:
2524
      # TODO: check OVS link
2525
      self.LogInfo("OVS links are currently not checked for correctness")
2526

    
2527
    if constants.INIC_MAC in params:
2528
      mac = params[constants.INIC_MAC]
2529
      if mac is None:
2530
        raise errors.OpPrereqError("Cannot unset the NIC MAC address",
2531
                                   errors.ECODE_INVAL)
2532
      elif mac in (constants.VALUE_AUTO, constants.VALUE_GENERATE):
2533
        # otherwise generate the MAC address
2534
        params[constants.INIC_MAC] = \
2535
          self.cfg.GenerateMAC(new_net_uuid, self.proc.GetECId())
2536
      else:
2537
        # or validate/reserve the current one
2538
        try:
2539
          self.cfg.ReserveMAC(mac, self.proc.GetECId())
2540
        except errors.ReservationError:
2541
          raise errors.OpPrereqError("MAC address '%s' already in use"
2542
                                     " in cluster" % mac,
2543
                                     errors.ECODE_NOTUNIQUE)
2544
    elif new_net_uuid != old_net_uuid:
2545

    
2546
      def get_net_prefix(net_uuid):
2547
        mac_prefix = None
2548
        if net_uuid:
2549
          nobj = self.cfg.GetNetwork(net_uuid)
2550
          mac_prefix = nobj.mac_prefix
2551

    
2552
        return mac_prefix
2553

    
2554
      new_prefix = get_net_prefix(new_net_uuid)
2555
      old_prefix = get_net_prefix(old_net_uuid)
2556
      if old_prefix != new_prefix:
2557
        params[constants.INIC_MAC] = \
2558
          self.cfg.GenerateMAC(new_net_uuid, self.proc.GetECId())
2559

    
2560
    # if there is a change in (ip, network) tuple
2561
    new_ip = params.get(constants.INIC_IP, old_ip)
2562
    if (new_ip, new_net_uuid) != (old_ip, old_net_uuid):
2563
      if new_ip:
2564
        # if IP is pool then require a network and generate one IP
2565
        if new_ip.lower() == constants.NIC_IP_POOL:
2566
          if new_net_uuid:
2567
            try:
2568
              new_ip = self.cfg.GenerateIp(new_net_uuid, self.proc.GetECId())
2569
            except errors.ReservationError:
2570
              raise errors.OpPrereqError("Unable to get a free IP"
2571
                                         " from the address pool",
2572
                                         errors.ECODE_STATE)
2573
            self.LogInfo("Chose IP %s from network %s",
2574
                         new_ip,
2575
                         new_net_obj.name)
2576
            params[constants.INIC_IP] = new_ip
2577
          else:
2578
            raise errors.OpPrereqError("ip=pool, but no network found",
2579
                                       errors.ECODE_INVAL)
2580
        # Reserve new IP if in the new network if any
2581
        elif new_net_uuid:
2582
          try:
2583
            self.cfg.ReserveIp(new_net_uuid, new_ip, self.proc.GetECId(),
2584
                               check=self.op.conflicts_check)
2585
            self.LogInfo("Reserving IP %s in network %s",
2586
                         new_ip, new_net_obj.name)
2587
          except errors.ReservationError:
2588
            raise errors.OpPrereqError("IP %s not available in network %s" %
2589
                                       (new_ip, new_net_obj.name),
2590
                                       errors.ECODE_NOTUNIQUE)
2591
        # new network is None so check if new IP is a conflicting IP
2592
        elif self.op.conflicts_check:
2593
          _CheckForConflictingIp(self, new_ip, pnode)
2594

    
2595
      # release old IP if old network is not None
2596
      if old_ip and old_net_uuid:
2597
        try:
2598
          self.cfg.ReleaseIp(old_net_uuid, old_ip, self.proc.GetECId())
2599
        except errors.AddressPoolError:
2600
          logging.warning("Release IP %s not contained in network %s",
2601
                          old_ip, old_net_obj.name)
2602

    
2603
    # there are no changes in (ip, network) tuple and old network is not None
2604
    elif (old_net_uuid is not None and
2605
          (req_link is not None or req_mode is not None)):
2606
      raise errors.OpPrereqError("Not allowed to change link or mode of"
2607
                                 " a NIC that is connected to a network",
2608
                                 errors.ECODE_INVAL)
2609

    
2610
    private.params = new_params
2611
    private.filled = new_filled_params
2612

    
2613
  def _PreCheckDiskTemplate(self, pnode_info):
2614
    """CheckPrereq checks related to a new disk template."""
2615
    # Arguments are passed to avoid configuration lookups
2616
    instance = self.instance
2617
    pnode = instance.primary_node
2618
    cluster = self.cluster
2619
    if instance.disk_template == self.op.disk_template:
2620
      raise errors.OpPrereqError("Instance already has disk template %s" %
2621
                                 instance.disk_template, errors.ECODE_INVAL)
2622

    
2623
    if (instance.disk_template,
2624
        self.op.disk_template) not in self._DISK_CONVERSIONS:
2625
      raise errors.OpPrereqError("Unsupported disk template conversion from"
2626
                                 " %s to %s" % (instance.disk_template,
2627
                                                self.op.disk_template),
2628
                                 errors.ECODE_INVAL)
2629
    CheckInstanceState(self, instance, INSTANCE_DOWN,
2630
                       msg="cannot change disk template")
2631
    if self.op.disk_template in constants.DTS_INT_MIRROR:
2632
      if self.op.remote_node == pnode:
2633
        raise errors.OpPrereqError("Given new secondary node %s is the same"
2634
                                   " as the primary node of the instance" %
2635
                                   self.op.remote_node, errors.ECODE_STATE)
2636
      CheckNodeOnline(self, self.op.remote_node)
2637
      CheckNodeNotDrained(self, self.op.remote_node)
2638
      # FIXME: here we assume that the old instance type is DT_PLAIN
2639
      assert instance.disk_template == constants.DT_PLAIN
2640
      disks = [{constants.IDISK_SIZE: d.size,
2641
                constants.IDISK_VG: d.logical_id[0]}
2642
               for d in instance.disks]
2643
      required = ComputeDiskSizePerVG(self.op.disk_template, disks)
2644
      CheckNodesFreeDiskPerVG(self, [self.op.remote_node], required)
2645

    
2646
      snode_info = self.cfg.GetNodeInfo(self.op.remote_node)
2647
      snode_group = self.cfg.GetNodeGroup(snode_info.group)
2648
      ipolicy = ganeti.masterd.instance.CalculateGroupIPolicy(cluster,
2649
                                                              snode_group)
2650
      CheckTargetNodeIPolicy(self, ipolicy, instance, snode_info, self.cfg,
2651
                             ignore=self.op.ignore_ipolicy)
2652
      if pnode_info.group != snode_info.group:
2653
        self.LogWarning("The primary and secondary nodes are in two"
2654
                        " different node groups; the disk parameters"
2655
                        " from the first disk's node group will be"
2656
                        " used")
2657

    
2658
    if not self.op.disk_template in constants.DTS_EXCL_STORAGE:
2659
      # Make sure none of the nodes require exclusive storage
2660
      nodes = [pnode_info]
2661
      if self.op.disk_template in constants.DTS_INT_MIRROR:
2662
        assert snode_info
2663
        nodes.append(snode_info)
2664
      has_es = lambda n: IsExclusiveStorageEnabledNode(self.cfg, n)
2665
      if compat.any(map(has_es, nodes)):
2666
        errmsg = ("Cannot convert disk template from %s to %s when exclusive"
2667
                  " storage is enabled" % (instance.disk_template,
2668
                                           self.op.disk_template))
2669
        raise errors.OpPrereqError(errmsg, errors.ECODE_STATE)
2670

    
2671
  def CheckPrereq(self): # pylint: disable=R0914
2672
    """Check prerequisites.
2673

2674
    This only checks the instance list against the existing names.
2675

2676
    """
2677
    assert self.op.instance_name in self.owned_locks(locking.LEVEL_INSTANCE)
2678
    instance = self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
2679

    
2680
    cluster = self.cluster = self.cfg.GetClusterInfo()
2681
    assert self.instance is not None, \
2682
      "Cannot retrieve locked instance %s" % self.op.instance_name
2683

    
2684
    pnode = instance.primary_node
2685

    
2686
    self.warn = []
2687

    
2688
    if (self.op.pnode is not None and self.op.pnode != pnode and
2689
        not self.op.force):
2690
      # verify that the instance is not up
2691
      instance_info = self.rpc.call_instance_info(pnode, instance.name,
2692
                                                  instance.hypervisor)
2693
      if instance_info.fail_msg:
2694
        self.warn.append("Can't get instance runtime information: %s" %
2695
                         instance_info.fail_msg)
2696
      elif instance_info.payload:
2697
        raise errors.OpPrereqError("Instance is still running on %s" % pnode,
2698
                                   errors.ECODE_STATE)
2699

    
2700
    assert pnode in self.owned_locks(locking.LEVEL_NODE)
2701
    nodelist = list(instance.all_nodes)
2702
    pnode_info = self.cfg.GetNodeInfo(pnode)
2703
    self.diskparams = self.cfg.GetInstanceDiskParams(instance)
2704

    
2705
    #_CheckInstanceNodeGroups(self.cfg, self.op.instance_name, owned_groups)
2706
    assert pnode_info.group in self.owned_locks(locking.LEVEL_NODEGROUP)
2707
    group_info = self.cfg.GetNodeGroup(pnode_info.group)
2708

    
2709
    # dictionary with instance information after the modification
2710
    ispec = {}
2711

    
2712
    if self.op.hotplug or self.op.hotplug_if_possible:
2713
      result = self.rpc.call_hotplug_supported(self.instance.primary_node,
2714
                                               self.instance)
2715
      if result.fail_msg:
2716
        if self.op.hotplug:
2717
          result.Raise("Hotplug is not possible: %s" % result.fail_msg,
2718
                       prereq=True)
2719
        else:
2720
          self.LogWarning(result.fail_msg)
2721
          self.op.hotplug = False
2722
          self.LogInfo("Modification will take place without hotplugging.")
2723
      else:
2724
        self.op.hotplug = True
2725

    
2726
    # Check disk modifications. This is done here and not in CheckArguments
2727
    # (as with NICs), because we need to know the instance's disk template
2728
    if instance.disk_template == constants.DT_EXT:
2729
      self._CheckMods("disk", self.op.disks, {},
2730
                      self._VerifyDiskModification)
2731
    else:
2732
      self._CheckMods("disk", self.op.disks, constants.IDISK_PARAMS_TYPES,
2733
                      self._VerifyDiskModification)
2734

    
2735
    # Prepare disk/NIC modifications
2736
    self.diskmod = _PrepareContainerMods(self.op.disks, None)
2737
    self.nicmod = _PrepareContainerMods(self.op.nics, _InstNicModPrivate)
2738

    
2739
    # Check the validity of the `provider' parameter
2740
    if instance.disk_template in constants.DT_EXT:
2741
      for mod in self.diskmod:
2742
        ext_provider = mod[2].get(constants.IDISK_PROVIDER, None)
2743
        if mod[0] == constants.DDM_ADD:
2744
          if ext_provider is None:
2745
            raise errors.OpPrereqError("Instance template is '%s' and parameter"
2746
                                       " '%s' missing, during disk add" %
2747
                                       (constants.DT_EXT,
2748
                                        constants.IDISK_PROVIDER),
2749
                                       errors.ECODE_NOENT)
2750
        elif mod[0] == constants.DDM_MODIFY:
2751
          if ext_provider:
2752
            raise errors.OpPrereqError("Parameter '%s' is invalid during disk"
2753
                                       " modification" %
2754
                                       constants.IDISK_PROVIDER,
2755
                                       errors.ECODE_INVAL)
2756
    else:
2757
      for mod in self.diskmod:
2758
        ext_provider = mod[2].get(constants.IDISK_PROVIDER, None)
2759
        if ext_provider is not None:
2760
          raise errors.OpPrereqError("Parameter '%s' is only valid for"
2761
                                     " instances of type '%s'" %
2762
                                     (constants.IDISK_PROVIDER,
2763
                                      constants.DT_EXT),
2764
                                     errors.ECODE_INVAL)
2765

    
2766
    # OS change
2767
    if self.op.os_name and not self.op.force:
2768
      CheckNodeHasOS(self, instance.primary_node, self.op.os_name,
2769
                     self.op.force_variant)
2770
      instance_os = self.op.os_name
2771
    else:
2772
      instance_os = instance.os
2773

    
2774
    assert not (self.op.disk_template and self.op.disks), \
2775
      "Can't modify disk template and apply disk changes at the same time"
2776

    
2777
    if self.op.disk_template:
2778
      self._PreCheckDiskTemplate(pnode_info)
2779

    
2780
    # hvparams processing
2781
    if self.op.hvparams:
2782
      hv_type = instance.hypervisor
2783
      i_hvdict = GetUpdatedParams(instance.hvparams, self.op.hvparams)
2784
      utils.ForceDictType(i_hvdict, constants.HVS_PARAMETER_TYPES)
2785
      hv_new = cluster.SimpleFillHV(hv_type, instance.os, i_hvdict)
2786

    
2787
      # local check
2788
      hypervisor.GetHypervisorClass(hv_type).CheckParameterSyntax(hv_new)
2789
      CheckHVParams(self, nodelist, instance.hypervisor, hv_new)
2790
      self.hv_proposed = self.hv_new = hv_new # the new actual values
2791
      self.hv_inst = i_hvdict # the new dict (without defaults)
2792
    else:
2793
      self.hv_proposed = cluster.SimpleFillHV(instance.hypervisor, instance.os,
2794
                                              instance.hvparams)
2795
      self.hv_new = self.hv_inst = {}
2796

    
2797
    # beparams processing
2798
    if self.op.beparams:
2799
      i_bedict = GetUpdatedParams(instance.beparams, self.op.beparams,
2800
                                  use_none=True)
2801
      objects.UpgradeBeParams(i_bedict)
2802
      utils.ForceDictType(i_bedict, constants.BES_PARAMETER_TYPES)
2803
      be_new = cluster.SimpleFillBE(i_bedict)
2804
      self.be_proposed = self.be_new = be_new # the new actual values
2805
      self.be_inst = i_bedict # the new dict (without defaults)
2806
    else:
2807
      self.be_new = self.be_inst = {}
2808
      self.be_proposed = cluster.SimpleFillBE(instance.beparams)
2809
    be_old = cluster.FillBE(instance)
2810

    
2811
    # CPU param validation -- checking every time a parameter is
2812
    # changed to cover all cases where either CPU mask or vcpus have
2813
    # changed
2814
    if (constants.BE_VCPUS in self.be_proposed and
2815
        constants.HV_CPU_MASK in self.hv_proposed):
2816
      cpu_list = \
2817
        utils.ParseMultiCpuMask(self.hv_proposed[constants.HV_CPU_MASK])
2818
      # Verify mask is consistent with number of vCPUs. Can skip this
2819
      # test if only 1 entry in the CPU mask, which means same mask
2820
      # is applied to all vCPUs.
2821
      if (len(cpu_list) > 1 and
2822
          len(cpu_list) != self.be_proposed[constants.BE_VCPUS]):
2823
        raise errors.OpPrereqError("Number of vCPUs [%d] does not match the"
2824
                                   " CPU mask [%s]" %
2825
                                   (self.be_proposed[constants.BE_VCPUS],
2826
                                    self.hv_proposed[constants.HV_CPU_MASK]),
2827
                                   errors.ECODE_INVAL)
2828

    
2829
      # Only perform this test if a new CPU mask is given
2830
      if constants.HV_CPU_MASK in self.hv_new:
2831
        # Calculate the largest CPU number requested
2832
        max_requested_cpu = max(map(max, cpu_list))
2833
        # Check that all of the instance's nodes have enough physical CPUs to
2834
        # satisfy the requested CPU mask
2835
        _CheckNodesPhysicalCPUs(self, instance.all_nodes,
2836
                                max_requested_cpu + 1, instance.hypervisor)
2837

    
2838
    # osparams processing
2839
    if self.op.osparams:
2840
      i_osdict = GetUpdatedParams(instance.osparams, self.op.osparams)
2841
      CheckOSParams(self, True, nodelist, instance_os, i_osdict)
2842
      self.os_inst = i_osdict # the new dict (without defaults)
2843
    else:
2844
      self.os_inst = {}
2845

    
2846
    #TODO(dynmem): do the appropriate check involving MINMEM
2847
    if (constants.BE_MAXMEM in self.op.beparams and not self.op.force and
2848
        be_new[constants.BE_MAXMEM] > be_old[constants.BE_MAXMEM]):
2849
      mem_check_list = [pnode]
2850
      if be_new[constants.BE_AUTO_BALANCE]:
2851
        # either we changed auto_balance to yes or it was from before
2852
        mem_check_list.extend(instance.secondary_nodes)
2853
      instance_info = self.rpc.call_instance_info(pnode, instance.name,
2854
                                                  instance.hypervisor)
2855
      nodeinfo = self.rpc.call_node_info(mem_check_list, None,
2856
                                         [instance.hypervisor], False)
2857
      pninfo = nodeinfo[pnode]
2858
      msg = pninfo.fail_msg
2859
      if msg:
2860
        # Assume the primary node is unreachable and go ahead
2861
        self.warn.append("Can't get info from primary node %s: %s" %
2862
                         (pnode, msg))
2863
      else:
2864
        (_, _, (pnhvinfo, )) = pninfo.payload
2865
        if not isinstance(pnhvinfo.get("memory_free", None), int):
2866
          self.warn.append("Node data from primary node %s doesn't contain"
2867
                           " free memory information" % pnode)
2868
        elif instance_info.fail_msg:
2869
          self.warn.append("Can't get instance runtime information: %s" %
2870
                           instance_info.fail_msg)
2871
        else:
2872
          if instance_info.payload:
2873
            current_mem = int(instance_info.payload["memory"])
2874
          else:
2875
            # Assume instance not running
2876
            # (there is a slight race condition here, but it's not very
2877
            # probable, and we have no other way to check)
2878
            # TODO: Describe race condition
2879
            current_mem = 0
2880
          #TODO(dynmem): do the appropriate check involving MINMEM
2881
          miss_mem = (be_new[constants.BE_MAXMEM] - current_mem -
2882
                      pnhvinfo["memory_free"])
2883
          if miss_mem > 0:
2884
            raise errors.OpPrereqError("This change will prevent the instance"
2885
                                       " from starting, due to %d MB of memory"
2886
                                       " missing on its primary node" %
2887
                                       miss_mem, errors.ECODE_NORES)
2888

    
2889
      if be_new[constants.BE_AUTO_BALANCE]:
2890
        for node, nres in nodeinfo.items():
2891
          if node not in instance.secondary_nodes:
2892
            continue
2893
          nres.Raise("Can't get info from secondary node %s" % node,
2894
                     prereq=True, ecode=errors.ECODE_STATE)
2895
          (_, _, (nhvinfo, )) = nres.payload
2896
          if not isinstance(nhvinfo.get("memory_free", None), int):
2897
            raise errors.OpPrereqError("Secondary node %s didn't return free"
2898
                                       " memory information" % node,
2899
                                       errors.ECODE_STATE)
2900
          #TODO(dynmem): do the appropriate check involving MINMEM
2901
          elif be_new[constants.BE_MAXMEM] > nhvinfo["memory_free"]:
2902
            raise errors.OpPrereqError("This change will prevent the instance"
2903
                                       " from failover to its secondary node"
2904
                                       " %s, due to not enough memory" % node,
2905
                                       errors.ECODE_STATE)
2906

    
2907
    if self.op.runtime_mem:
2908
      remote_info = self.rpc.call_instance_info(instance.primary_node,
2909
                                                instance.name,
2910
                                                instance.hypervisor)
2911
      remote_info.Raise("Error checking node %s" % instance.primary_node)
2912
      if not remote_info.payload: # not running already
2913
        raise errors.OpPrereqError("Instance %s is not running" %
2914
                                   instance.name, errors.ECODE_STATE)
2915

    
2916
      current_memory = remote_info.payload["memory"]
2917
      if (not self.op.force and
2918
           (self.op.runtime_mem > self.be_proposed[constants.BE_MAXMEM] or
2919
            self.op.runtime_mem < self.be_proposed[constants.BE_MINMEM])):
2920
        raise errors.OpPrereqError("Instance %s must have memory between %d"
2921
                                   " and %d MB of memory unless --force is"
2922
                                   " given" %
2923
                                   (instance.name,
2924
                                    self.be_proposed[constants.BE_MINMEM],
2925
                                    self.be_proposed[constants.BE_MAXMEM]),
2926
                                   errors.ECODE_INVAL)
2927

    
2928
      delta = self.op.runtime_mem - current_memory
2929
      if delta > 0:
2930
        CheckNodeFreeMemory(self, instance.primary_node,
2931
                            "ballooning memory for instance %s" %
2932
                            instance.name, delta, instance.hypervisor)
2933

    
2934
    if self.op.disks and instance.disk_template == constants.DT_DISKLESS:
2935
      raise errors.OpPrereqError("Disk operations not supported for"
2936
                                 " diskless instances", errors.ECODE_INVAL)
2937

    
2938
    def _PrepareNicCreate(_, params, private):
2939
      self._PrepareNicModification(params, private, None, None,
2940
                                   {}, cluster, pnode)
2941
      return (None, None)
2942

    
2943
    def _PrepareNicMod(_, nic, params, private):
2944
      self._PrepareNicModification(params, private, nic.ip, nic.network,
2945
                                   nic.nicparams, cluster, pnode)
2946
      return None
2947

    
2948
    def _PrepareNicRemove(_, params, __):
2949
      ip = params.ip
2950
      net = params.network
2951
      if net is not None and ip is not None:
2952
        self.cfg.ReleaseIp(net, ip, self.proc.GetECId())
2953

    
2954
    # Verify NIC changes (operating on copy)
2955
    nics = instance.nics[:]
2956
    _ApplyContainerMods("NIC", nics, None, self.nicmod,
2957
                        _PrepareNicCreate, _PrepareNicMod, _PrepareNicRemove)
2958
    if len(nics) > constants.MAX_NICS:
2959
      raise errors.OpPrereqError("Instance has too many network interfaces"
2960
                                 " (%d), cannot add more" % constants.MAX_NICS,
2961
                                 errors.ECODE_STATE)
2962

    
2963
    def _PrepareDiskMod(_, disk, params, __):
2964
      disk.name = params.get(constants.IDISK_NAME, None)
2965

    
2966
    # Verify disk changes (operating on a copy)
2967
    disks = copy.deepcopy(instance.disks)
2968
    _ApplyContainerMods("disk", disks, None, self.diskmod, None,
2969
                        _PrepareDiskMod, None)
2970
    utils.ValidateDeviceNames("disk", disks)
2971
    if len(disks) > constants.MAX_DISKS:
2972
      raise errors.OpPrereqError("Instance has too many disks (%d), cannot add"
2973
                                 " more" % constants.MAX_DISKS,
2974
                                 errors.ECODE_STATE)
2975
    disk_sizes = [disk.size for disk in instance.disks]
2976
    disk_sizes.extend(params["size"] for (op, idx, params, private) in
2977
                      self.diskmod if op == constants.DDM_ADD)
2978
    ispec[constants.ISPEC_DISK_COUNT] = len(disk_sizes)
2979
    ispec[constants.ISPEC_DISK_SIZE] = disk_sizes
2980

    
2981
    if self.op.offline is not None and self.op.offline:
2982
      CheckInstanceState(self, instance, CAN_CHANGE_INSTANCE_OFFLINE,
2983
                         msg="can't change to offline")
2984

    
2985
    # Pre-compute NIC changes (necessary to use result in hooks)
2986
    self._nic_chgdesc = []
2987
    if self.nicmod:
2988
      # Operate on copies as this is still in prereq
2989
      nics = [nic.Copy() for nic in instance.nics]
2990
      _ApplyContainerMods("NIC", nics, self._nic_chgdesc, self.nicmod,
2991
                          self._CreateNewNic, self._ApplyNicMods,
2992
                          self._RemoveNic)
2993
      # Verify that NIC names are unique and valid
2994
      utils.ValidateDeviceNames("NIC", nics)
2995
      self._new_nics = nics
2996
      ispec[constants.ISPEC_NIC_COUNT] = len(self._new_nics)
2997
    else:
2998
      self._new_nics = None
2999
      ispec[constants.ISPEC_NIC_COUNT] = len(instance.nics)
3000

    
3001
    if not self.op.ignore_ipolicy:
3002
      ipolicy = ganeti.masterd.instance.CalculateGroupIPolicy(cluster,
3003
                                                              group_info)
3004

    
3005
      # Fill ispec with backend parameters
3006
      ispec[constants.ISPEC_SPINDLE_USE] = \
3007
        self.be_new.get(constants.BE_SPINDLE_USE, None)
3008
      ispec[constants.ISPEC_CPU_COUNT] = self.be_new.get(constants.BE_VCPUS,
3009
                                                         None)
3010

    
3011
      # Copy ispec to verify parameters with min/max values separately
3012
      if self.op.disk_template:
3013
        new_disk_template = self.op.disk_template
3014
      else:
3015
        new_disk_template = instance.disk_template
3016
      ispec_max = ispec.copy()
3017
      ispec_max[constants.ISPEC_MEM_SIZE] = \
3018
        self.be_new.get(constants.BE_MAXMEM, None)
3019
      res_max = _ComputeIPolicyInstanceSpecViolation(ipolicy, ispec_max,
3020
                                                     new_disk_template)
3021
      ispec_min = ispec.copy()
3022
      ispec_min[constants.ISPEC_MEM_SIZE] = \
3023
        self.be_new.get(constants.BE_MINMEM, None)
3024
      res_min = _ComputeIPolicyInstanceSpecViolation(ipolicy, ispec_min,
3025
                                                     new_disk_template)
3026

    
3027
      if (res_max or res_min):
3028
        # FIXME: Improve error message by including information about whether
3029
        # the upper or lower limit of the parameter fails the ipolicy.
3030
        msg = ("Instance allocation to group %s (%s) violates policy: %s" %
3031
               (group_info, group_info.name,
3032
                utils.CommaJoin(set(res_max + res_min))))
3033
        raise errors.OpPrereqError(msg, errors.ECODE_INVAL)
3034

    
3035
  def _ConvertPlainToDrbd(self, feedback_fn):
3036
    """Converts an instance from plain to drbd.
3037

3038
    """
3039
    feedback_fn("Converting template to drbd")
3040
    instance = self.instance
3041
    pnode = instance.primary_node
3042
    snode = self.op.remote_node
3043

    
3044
    assert instance.disk_template == constants.DT_PLAIN
3045

    
3046
    # create a fake disk info for _GenerateDiskTemplate
3047
    disk_info = [{constants.IDISK_SIZE: d.size, constants.IDISK_MODE: d.mode,
3048
                  constants.IDISK_VG: d.logical_id[0],
3049
                  constants.IDISK_NAME: d.name}
3050
                 for d in instance.disks]
3051
    new_disks = GenerateDiskTemplate(self, self.op.disk_template,
3052
                                     instance.name, pnode, [snode],
3053
                                     disk_info, None, None, 0, feedback_fn,
3054
                                     self.diskparams)
3055
    anno_disks = rpc.AnnotateDiskParams(constants.DT_DRBD8, new_disks,
3056
                                        self.diskparams)
3057
    p_excl_stor = IsExclusiveStorageEnabledNodeName(self.cfg, pnode)
3058
    s_excl_stor = IsExclusiveStorageEnabledNodeName(self.cfg, snode)
3059
    info = GetInstanceInfoText(instance)
3060
    feedback_fn("Creating additional volumes...")
3061
    # first, create the missing data and meta devices
3062
    for disk in anno_disks:
3063
      # unfortunately this is... not too nice
3064
      CreateSingleBlockDev(self, pnode, instance, disk.children[1],
3065
                           info, True, p_excl_stor)
3066
      for child in disk.children:
3067
        CreateSingleBlockDev(self, snode, instance, child, info, True,
3068
                             s_excl_stor)
3069
    # at this stage, all new LVs have been created, we can rename the
3070
    # old ones
3071
    feedback_fn("Renaming original volumes...")
3072
    rename_list = [(o, n.children[0].logical_id)
3073
                   for (o, n) in zip(instance.disks, new_disks)]
3074
    result = self.rpc.call_blockdev_rename(pnode, rename_list)
3075
    result.Raise("Failed to rename original LVs")
3076

    
3077
    feedback_fn("Initializing DRBD devices...")
3078
    # all child devices are in place, we can now create the DRBD devices
3079
    try:
3080
      for disk in anno_disks:
3081
        for (node, excl_stor) in [(pnode, p_excl_stor), (snode, s_excl_stor)]:
3082
          f_create = node == pnode
3083
          CreateSingleBlockDev(self, node, instance, disk, info, f_create,
3084
                               excl_stor)
3085
    except errors.GenericError, e:
3086
      feedback_fn("Initializing of DRBD devices failed;"
3087
                  " renaming back original volumes...")
3088
      for disk in new_disks:
3089
        self.cfg.SetDiskID(disk, pnode)
3090
      rename_back_list = [(n.children[0], o.logical_id)
3091
                          for (n, o) in zip(new_disks, instance.disks)]
3092
      result = self.rpc.call_blockdev_rename(pnode, rename_back_list)
3093
      result.Raise("Failed to rename LVs back after error %s" % str(e))
3094
      raise
3095

    
3096
    # at this point, the instance has been modified
3097
    instance.disk_template = constants.DT_DRBD8
3098
    instance.disks = new_disks
3099
    self.cfg.Update(instance, feedback_fn)
3100

    
3101
    # Release node locks while waiting for sync
3102
    ReleaseLocks(self, locking.LEVEL_NODE)
3103

    
3104
    # disks are created, waiting for sync
3105
    disk_abort = not WaitForSync(self, instance,
3106
                                 oneshot=not self.op.wait_for_sync)
3107
    if disk_abort:
3108
      raise errors.OpExecError("There are some degraded disks for"
3109
                               " this instance, please cleanup manually")
3110

    
3111
    # Node resource locks will be released by caller
3112

    
3113
  def _ConvertDrbdToPlain(self, feedback_fn):
3114
    """Converts an instance from drbd to plain.
3115

3116
    """
3117
    instance = self.instance
3118

    
3119
    assert len(instance.secondary_nodes) == 1
3120
    assert instance.disk_template == constants.DT_DRBD8
3121

    
3122
    pnode = instance.primary_node
3123
    snode = instance.secondary_nodes[0]
3124
    feedback_fn("Converting template to plain")
3125

    
3126
    old_disks = AnnotateDiskParams(instance, instance.disks, self.cfg)
3127
    new_disks = [d.children[0] for d in instance.disks]
3128

    
3129
    # copy over size, mode and name
3130
    for parent, child in zip(old_disks, new_disks):
3131
      child.size = parent.size
3132
      child.mode = parent.mode
3133
      child.name = parent.name
3134

    
3135
    # this is a DRBD disk, return its port to the pool
3136
    # NOTE: this must be done right before the call to cfg.Update!
3137
    for disk in old_disks:
3138
      tcp_port = disk.logical_id[2]
3139
      self.cfg.AddTcpUdpPort(tcp_port)
3140

    
3141
    # update instance structure
3142
    instance.disks = new_disks
3143
    instance.disk_template = constants.DT_PLAIN
3144
    _UpdateIvNames(0, instance.disks)
3145
    self.cfg.Update(instance, feedback_fn)
3146

    
3147
    # Release locks in case removing disks takes a while
3148
    ReleaseLocks(self, locking.LEVEL_NODE)
3149

    
3150
    feedback_fn("Removing volumes on the secondary node...")
3151
    for disk in old_disks:
3152
      self.cfg.SetDiskID(disk, snode)
3153
      msg = self.rpc.call_blockdev_remove(snode, disk).fail_msg
3154
      if msg:
3155
        self.LogWarning("Could not remove block device %s on node %s,"
3156
                        " continuing anyway: %s", disk.iv_name, snode, msg)
3157

    
3158
    feedback_fn("Removing unneeded volumes on the primary node...")
3159
    for idx, disk in enumerate(old_disks):
3160
      meta = disk.children[1]
3161
      self.cfg.SetDiskID(meta, pnode)
3162
      msg = self.rpc.call_blockdev_remove(pnode, meta).fail_msg
3163
      if msg:
3164
        self.LogWarning("Could not remove metadata for disk %d on node %s,"
3165
                        " continuing anyway: %s", idx, pnode, msg)
3166

    
3167
  def _HotplugDevice(self, action, dev_type, device, extra, seq):
3168
    self.LogInfo("Trying to hotplug device...")
3169
    result = self.rpc.call_hotplug_device(self.instance.primary_node,
3170
                                          self.instance, action, dev_type,
3171
                                          (device, self.instance),
3172
                                          extra, seq)
3173
    if result.fail_msg:
3174
      self.LogWarning("Could not hotplug device: %s" % result.fail_msg)
3175
      self.LogInfo("Continuing execution..")
3176
    else:
3177
      self.LogInfo("Hotplug done.")
3178

    
3179
  def _CreateNewDisk(self, idx, params, _):
3180
    """Creates a new disk.
3181

3182
    """
3183
    instance = self.instance
3184

    
3185
    # add a new disk
3186
    if instance.disk_template in constants.DTS_FILEBASED:
3187
      (file_driver, file_path) = instance.disks[0].logical_id
3188
      file_path = os.path.dirname(file_path)
3189
    else:
3190
      file_driver = file_path = None
3191

    
3192
    disk = \
3193
      GenerateDiskTemplate(self, instance.disk_template, instance.name,
3194
                           instance.primary_node, instance.secondary_nodes,
3195
                           [params], file_path, file_driver, idx,
3196
                           self.Log, self.diskparams)[0]
3197

    
3198
    new_disks = CreateDisks(self, instance, disks=[disk])
3199

    
3200
    if self.cluster.prealloc_wipe_disks:
3201
      # Wipe new disk
3202
      WipeOrCleanupDisks(self, instance,
3203
                         disks=[(idx, disk, 0)],
3204
                         cleanup=new_disks)
3205

    
3206
    if self.op.hotplug:
3207
      # _, device_info = AssembleInstanceDisks(self, self.instance,
3208
      #                                       [disk], check=False)
3209
      self.cfg.SetDiskID(disk, self.instance.primary_node)
3210
      result = self.rpc.call_blockdev_assemble(self.instance.primary_node,
3211
                                               (disk, self.instance),
3212
                                               self.instance.name, True, idx)
3213
      if result.fail_msg:
3214
        self.LogWarning("Can't assemble newly created disk %d: %s",
3215
                        idx, result.fail_msg)
3216
      else:
3217
        # _, _, dev_path = device_info[0]
3218
        _, link_name = result.payload
3219
        self._HotplugDevice(constants.HOTPLUG_ACTION_ADD,
3220
                            constants.HOTPLUG_TARGET_DISK,
3221
                            disk, link_name, idx)
3222

    
3223
    return (disk, [
3224
      ("disk/%d" % idx, "add:size=%s,mode=%s" % (disk.size, disk.mode)),
3225
      ])
3226

    
3227
  def _ModifyDisk(self, idx, disk, params, _):
3228
    """Modifies a disk.
3229

3230
    """
3231
    changes = []
3232
    for key, value in params.iteritems():
3233
      if key in [constants.IDISK_MODE, constants.IDISK_NAME]:
3234
        setattr(disk, key, value)
3235
        changes.append(("disk.%s/%d" % (key, idx), value))
3236
      elif self.instance.disk_template == constants.DT_EXT:
3237
        # stolen from GetUpdatedParams: default means reset/delete
3238
        if value.lower() == constants.VALUE_DEFAULT:
3239
          try:
3240
            del disk.params[key]
3241
          except KeyError:
3242
            pass
3243
        else:
3244
          disk.params[key] = value
3245
        changes.append(("disk.params:%s/%d" % (key, idx), value))
3246

    
3247
    return changes
3248

    
3249
  def _RemoveDisk(self, idx, root, _):
3250
    """Removes a disk.
3251

3252
    """
3253
    if self.op.hotplug:
3254
      self._HotplugDevice(constants.HOTPLUG_ACTION_REMOVE,
3255
                          constants.HOTPLUG_TARGET_DISK,
3256
                          root, None, idx)
3257
      ShutdownInstanceDisks(self, self.instance, [root])
3258

    
3259
    (anno_disk,) = AnnotateDiskParams(self.instance, [root], self.cfg)
3260
    for node, disk in anno_disk.ComputeNodeTree(self.instance.primary_node):
3261
      if self.op.keep_disks and disk.dev_type in constants.DT_EXT:
3262
        continue
3263
      self.cfg.SetDiskID(disk, node)
3264
      msg = self.rpc.call_blockdev_remove(node, disk).fail_msg
3265
      if msg:
3266
        self.LogWarning("Could not remove disk/%d on node '%s': %s,"
3267
                        " continuing anyway", idx, node, msg)
3268

    
3269
    # if this is a DRBD disk, return its port to the pool
3270
    if root.dev_type in constants.LDS_DRBD:
3271
      self.cfg.AddTcpUdpPort(root.logical_id[2])
3272

    
3273
  def _CreateNewNic(self, idx, params, private):
3274
    """Creates data structure for a new network interface.
3275

3276
    """
3277
    mac = params[constants.INIC_MAC]
3278
    ip = params.get(constants.INIC_IP, None)
3279
    net = params.get(constants.INIC_NETWORK, None)
3280
    name = params.get(constants.INIC_NAME, None)
3281
    net_uuid = self.cfg.LookupNetwork(net)
3282
    #TODO: not private.filled?? can a nic have no nicparams??
3283
    nicparams = private.filled
3284
    nobj = objects.NIC(mac=mac, ip=ip, network=net_uuid, name=name,
3285
                       nicparams=nicparams)
3286
    nobj.uuid = self.cfg.GenerateUniqueID(self.proc.GetECId())
3287

    
3288
    if self.op.hotplug:
3289
      self._HotplugDevice(constants.HOTPLUG_ACTION_ADD,
3290
                          constants.HOTPLUG_TARGET_NIC,
3291
                          nobj, None, idx)
3292

    
3293
    desc = [
3294
      ("nic.%d" % idx,
3295
       "add:mac=%s,ip=%s,mode=%s,link=%s,network=%s" %
3296
       (mac, ip, private.filled[constants.NIC_MODE],
3297
       private.filled[constants.NIC_LINK], net)),
3298
      ]
3299

    
3300
    return (nobj, desc)
3301

    
3302
  def _ApplyNicMods(self, idx, nic, params, private):
3303
    """Modifies a network interface.
3304

3305
    """
3306
    changes = []
3307

    
3308
    for key in [constants.INIC_MAC, constants.INIC_IP, constants.INIC_NAME]:
3309
      if key in params:
3310
        changes.append(("nic.%s/%d" % (key, idx), params[key]))
3311
        setattr(nic, key, params[key])
3312

    
3313
    new_net = params.get(constants.INIC_NETWORK, nic.network)
3314
    new_net_uuid = self.cfg.LookupNetwork(new_net)
3315
    if new_net_uuid != nic.network:
3316
      changes.append(("nic.network/%d" % idx, new_net))
3317
      nic.network = new_net_uuid
3318

    
3319
    if private.filled:
3320
      nic.nicparams = private.filled
3321

    
3322
      for (key, val) in nic.nicparams.items():
3323
        changes.append(("nic.%s/%d" % (key, idx), val))
3324

    
3325
    if self.op.hotplug:
3326
      self._HotplugDevice(constants.HOTPLUG_ACTION_MODIFY,
3327
                          constants.HOTPLUG_TARGET_NIC,
3328
                          nic, None, idx)
3329

    
3330
    return changes
3331

    
3332
  def _RemoveNic(self, idx, nic, _):
3333
    if self.op.hotplug:
3334
      self._HotplugDevice(constants.HOTPLUG_ACTION_REMOVE,
3335
                          constants.HOTPLUG_TARGET_NIC,
3336
                          nic, None, idx)
3337

    
3338
  def Exec(self, feedback_fn):
3339
    """Modifies an instance.
3340

3341
    All parameters take effect only at the next restart of the instance.
3342

3343
    """
3344
    # Process here the warnings from CheckPrereq, as we don't have a
3345
    # feedback_fn there.
3346
    # TODO: Replace with self.LogWarning
3347
    for warn in self.warn:
3348
      feedback_fn("WARNING: %s" % warn)
3349

    
3350
    assert ((self.op.disk_template is None) ^
3351
            bool(self.owned_locks(locking.LEVEL_NODE_RES))), \
3352
      "Not owning any node resource locks"
3353

    
3354
    result = []
3355
    instance = self.instance
3356

    
3357
    # New primary node
3358
    if self.op.pnode:
3359
      instance.primary_node = self.op.pnode
3360

    
3361
    # runtime memory
3362
    if self.op.runtime_mem:
3363
      rpcres = self.rpc.call_instance_balloon_memory(instance.primary_node,
3364
                                                     instance,
3365
                                                     self.op.runtime_mem)
3366
      rpcres.Raise("Cannot modify instance runtime memory")
3367
      result.append(("runtime_memory", self.op.runtime_mem))
3368

    
3369
    # Apply disk changes
3370
    _ApplyContainerMods("disk", instance.disks, result, self.diskmod,
3371
                        self._CreateNewDisk, self._ModifyDisk,
3372
                        self._RemoveDisk)
3373
    _UpdateIvNames(0, instance.disks)
3374

    
3375
    if self.op.disk_template:
3376
      if __debug__:
3377
        check_nodes = set(instance.all_nodes)
3378
        if self.op.remote_node:
3379
          check_nodes.add(self.op.remote_node)
3380
        for level in [locking.LEVEL_NODE, locking.LEVEL_NODE_RES]:
3381
          owned = self.owned_locks(level)
3382
          assert not (check_nodes - owned), \
3383
            ("Not owning the correct locks, owning %r, expected at least %r" %
3384
             (owned, check_nodes))
3385

    
3386
      r_shut = ShutdownInstanceDisks(self, instance)
3387
      if not r_shut:
3388
        raise errors.OpExecError("Cannot shutdown instance disks, unable to"
3389
                                 " proceed with disk template conversion")
3390
      mode = (instance.disk_template, self.op.disk_template)
3391
      try:
3392
        self._DISK_CONVERSIONS[mode](self, feedback_fn)
3393
      except:
3394
        self.cfg.ReleaseDRBDMinors(instance.name)
3395
        raise
3396
      result.append(("disk_template", self.op.disk_template))
3397

    
3398
      assert instance.disk_template == self.op.disk_template, \
3399
        ("Expected disk template '%s', found '%s'" %
3400
         (self.op.disk_template, instance.disk_template))
3401

    
3402
    # Release node and resource locks if there are any (they might already have
3403
    # been released during disk conversion)
3404
    ReleaseLocks(self, locking.LEVEL_NODE)
3405
    ReleaseLocks(self, locking.LEVEL_NODE_RES)
3406

    
3407
    # Apply NIC changes
3408
    if self._new_nics is not None:
3409
      instance.nics = self._new_nics
3410
      result.extend(self._nic_chgdesc)
3411

    
3412
    # hvparams changes
3413
    if self.op.hvparams:
3414
      instance.hvparams = self.hv_inst
3415
      for key, val in self.op.hvparams.iteritems():
3416
        result.append(("hv/%s" % key, val))
3417

    
3418
    # beparams changes
3419
    if self.op.beparams:
3420
      instance.beparams = self.be_inst
3421
      for key, val in self.op.beparams.iteritems():
3422
        result.append(("be/%s" % key, val))
3423

    
3424
    # OS change
3425
    if self.op.os_name:
3426
      instance.os = self.op.os_name
3427

    
3428
    # osparams changes
3429
    if self.op.osparams:
3430
      instance.osparams = self.os_inst
3431
      for key, val in self.op.osparams.iteritems():
3432
        result.append(("os/%s" % key, val))
3433

    
3434
    if self.op.offline is None:
3435
      # Ignore
3436
      pass
3437
    elif self.op.offline:
3438
      # Mark instance as offline
3439
      self.cfg.MarkInstanceOffline(instance.name)
3440
      result.append(("admin_state", constants.ADMINST_OFFLINE))
3441
    else:
3442
      # Mark instance as online, but stopped
3443
      self.cfg.MarkInstanceDown(instance.name)
3444
      result.append(("admin_state", constants.ADMINST_DOWN))
3445

    
3446
    self.cfg.Update(instance, feedback_fn, self.proc.GetECId())
3447

    
3448
    assert not (self.owned_locks(locking.LEVEL_NODE_RES) or
3449
                self.owned_locks(locking.LEVEL_NODE)), \
3450
      "All node locks should have been released by now"
3451

    
3452
    return result
3453

    
3454
  _DISK_CONVERSIONS = {
3455
    (constants.DT_PLAIN, constants.DT_DRBD8): _ConvertPlainToDrbd,
3456
    (constants.DT_DRBD8, constants.DT_PLAIN): _ConvertDrbdToPlain,
3457
    }
3458

    
3459

    
3460
class LUInstanceChangeGroup(LogicalUnit):
3461
  HPATH = "instance-change-group"
3462
  HTYPE = constants.HTYPE_INSTANCE
3463
  REQ_BGL = False
3464

    
3465
  def ExpandNames(self):
3466
    self.share_locks = ShareAll()
3467

    
3468
    self.needed_locks = {
3469
      locking.LEVEL_NODEGROUP: [],
3470
      locking.LEVEL_NODE: [],
3471
      locking.LEVEL_NODE_ALLOC: locking.ALL_SET,
3472
      }
3473

    
3474
    self._ExpandAndLockInstance()
3475

    
3476
    if self.op.target_groups:
3477
      self.req_target_uuids = map(self.cfg.LookupNodeGroup,
3478
                                  self.op.target_groups)
3479
    else:
3480
      self.req_target_uuids = None
3481

    
3482
    self.op.iallocator = GetDefaultIAllocator(self.cfg, self.op.iallocator)
3483

    
3484
  def DeclareLocks(self, level):
3485
    if level == locking.LEVEL_NODEGROUP:
3486
      assert not self.needed_locks[locking.LEVEL_NODEGROUP]
3487

    
3488
      if self.req_target_uuids:
3489
        lock_groups = set(self.req_target_uuids)
3490

    
3491
        # Lock all groups used by instance optimistically; this requires going
3492
        # via the node before it's locked, requiring verification later on
3493
        instance_groups = self.cfg.GetInstanceNodeGroups(self.op.instance_name)
3494
        lock_groups.update(instance_groups)
3495
      else:
3496
        # No target groups, need to lock all of them
3497
        lock_groups = locking.ALL_SET
3498

    
3499
      self.needed_locks[locking.LEVEL_NODEGROUP] = lock_groups
3500

    
3501
    elif level == locking.LEVEL_NODE:
3502
      if self.req_target_uuids:
3503
        # Lock all nodes used by instances
3504
        self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_APPEND
3505
        self._LockInstancesNodes()
3506

    
3507
        # Lock all nodes in all potential target groups
3508
        lock_groups = (frozenset(self.owned_locks(locking.LEVEL_NODEGROUP)) -
3509
                       self.cfg.GetInstanceNodeGroups(self.op.instance_name))
3510
        member_nodes = [node_name
3511
                        for group in lock_groups
3512
                        for node_name in self.cfg.GetNodeGroup(group).members]
3513
        self.needed_locks[locking.LEVEL_NODE].extend(member_nodes)
3514
      else:
3515
        # Lock all nodes as all groups are potential targets
3516
        self.needed_locks[locking.LEVEL_NODE] = locking.ALL_SET
3517

    
3518
  def CheckPrereq(self):
3519
    owned_instances = frozenset(self.owned_locks(locking.LEVEL_INSTANCE))
3520
    owned_groups = frozenset(self.owned_locks(locking.LEVEL_NODEGROUP))
3521
    owned_nodes = frozenset(self.owned_locks(locking.LEVEL_NODE))
3522

    
3523
    assert (self.req_target_uuids is None or
3524
            owned_groups.issuperset(self.req_target_uuids))
3525
    assert owned_instances == set([self.op.instance_name])
3526

    
3527
    # Get instance information
3528
    self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
3529

    
3530
    # Check if node groups for locked instance are still correct
3531
    assert owned_nodes.issuperset(self.instance.all_nodes), \
3532
      ("Instance %s's nodes changed while we kept the lock" %
3533
       self.op.instance_name)
3534

    
3535
    inst_groups = CheckInstanceNodeGroups(self.cfg, self.op.instance_name,
3536
                                          owned_groups)
3537

    
3538
    if self.req_target_uuids:
3539
      # User requested specific target groups
3540
      self.target_uuids = frozenset(self.req_target_uuids)
3541
    else:
3542
      # All groups except those used by the instance are potential targets
3543
      self.target_uuids = owned_groups - inst_groups
3544

    
3545
    conflicting_groups = self.target_uuids & inst_groups
3546
    if conflicting_groups:
3547
      raise errors.OpPrereqError("Can't use group(s) '%s' as targets, they are"
3548
                                 " used by the instance '%s'" %
3549
                                 (utils.CommaJoin(conflicting_groups),
3550
                                  self.op.instance_name),
3551
                                 errors.ECODE_INVAL)
3552

    
3553
    if not self.target_uuids:
3554
      raise errors.OpPrereqError("There are no possible target groups",
3555
                                 errors.ECODE_INVAL)
3556

    
3557
  def BuildHooksEnv(self):
3558
    """Build hooks env.
3559

3560
    """
3561
    assert self.target_uuids
3562

    
3563
    env = {
3564
      "TARGET_GROUPS": " ".join(self.target_uuids),
3565
      }
3566

    
3567
    env.update(BuildInstanceHookEnvByObject(self, self.instance))
3568

    
3569
    return env
3570

    
3571
  def BuildHooksNodes(self):
3572
    """Build hooks nodes.
3573

3574
    """
3575
    mn = self.cfg.GetMasterNode()
3576
    return ([mn], [mn])
3577

    
3578
  def Exec(self, feedback_fn):
3579
    instances = list(self.owned_locks(locking.LEVEL_INSTANCE))
3580

    
3581
    assert instances == [self.op.instance_name], "Instance not locked"
3582

    
3583
    req = iallocator.IAReqGroupChange(instances=instances,
3584
                                      target_groups=list(self.target_uuids))
3585
    ial = iallocator.IAllocator(self.cfg, self.rpc, req)
3586

    
3587
    ial.Run(self.op.iallocator)
3588

    
3589
    if not ial.success:
3590
      raise errors.OpPrereqError("Can't compute solution for changing group of"
3591
                                 " instance '%s' using iallocator '%s': %s" %
3592
                                 (self.op.instance_name, self.op.iallocator,
3593
                                  ial.info), errors.ECODE_NORES)
3594

    
3595
    jobs = LoadNodeEvacResult(self, ial.result, self.op.early_release, False)
3596

    
3597
    self.LogInfo("Iallocator returned %s job(s) for changing group of"
3598
                 " instance '%s'", len(jobs), self.op.instance_name)
3599

    
3600
    return ResultWithJobs(jobs)