4 # Copyright (C) 2006, 2007, 2008 Google Inc.
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.
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.
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
22 """Module implementing the master-side code."""
24 # pylint: disable-msg=W0613,W0201
34 from ganeti import ssh
35 from ganeti import utils
36 from ganeti import errors
37 from ganeti import hypervisor
38 from ganeti import locking
39 from ganeti import constants
40 from ganeti import objects
41 from ganeti import serializer
42 from ganeti import ssconf
45 class LogicalUnit(object):
46 """Logical Unit base class.
48 Subclasses must follow these rules:
49 - implement ExpandNames
50 - implement CheckPrereq (except when tasklets are used)
51 - implement Exec (except when tasklets are used)
52 - implement BuildHooksEnv
53 - redefine HPATH and HTYPE
54 - optionally redefine their run requirements:
55 REQ_BGL: the LU needs to hold the Big Ganeti Lock exclusively
57 Note that all commands require root permissions.
59 @ivar dry_run_result: the value (if any) that will be returned to the caller
60 in dry-run mode (signalled by opcode dry_run parameter)
68 def __init__(self, processor, op, context, rpc):
69 """Constructor for LogicalUnit.
71 This needs to be overridden in derived classes in order to check op
77 self.cfg = context.cfg
78 self.context = context
80 # Dicts used to declare locking needs to mcpu
81 self.needed_locks = None
82 self.acquired_locks = {}
83 self.share_locks = dict.fromkeys(locking.LEVELS, 0)
85 self.remove_locks = {}
86 # Used to force good behavior when calling helper functions
87 self.recalculate_locks = {}
90 self.LogWarning = processor.LogWarning
91 self.LogInfo = processor.LogInfo
92 self.LogStep = processor.LogStep
94 self.dry_run_result = None
99 for attr_name in self._OP_REQP:
100 attr_val = getattr(op, attr_name, None)
102 raise errors.OpPrereqError("Required parameter '%s' missing" %
105 self.CheckArguments()
108 """Returns the SshRunner object
112 self.__ssh = ssh.SshRunner(self.cfg.GetClusterName())
115 ssh = property(fget=__GetSSH)
117 def CheckArguments(self):
118 """Check syntactic validity for the opcode arguments.
120 This method is for doing a simple syntactic check and ensure
121 validity of opcode parameters, without any cluster-related
122 checks. While the same can be accomplished in ExpandNames and/or
123 CheckPrereq, doing these separate is better because:
125 - ExpandNames is left as as purely a lock-related function
126 - CheckPrereq is run after we have acquired locks (and possible
129 The function is allowed to change the self.op attribute so that
130 later methods can no longer worry about missing parameters.
135 def ExpandNames(self):
136 """Expand names for this LU.
138 This method is called before starting to execute the opcode, and it should
139 update all the parameters of the opcode to their canonical form (e.g. a
140 short node name must be fully expanded after this method has successfully
141 completed). This way locking, hooks, logging, ecc. can work correctly.
143 LUs which implement this method must also populate the self.needed_locks
144 member, as a dict with lock levels as keys, and a list of needed lock names
147 - use an empty dict if you don't need any lock
148 - if you don't need any lock at a particular level omit that level
149 - don't put anything for the BGL level
150 - if you want all locks at a level use locking.ALL_SET as a value
152 If you need to share locks (rather than acquire them exclusively) at one
153 level you can modify self.share_locks, setting a true value (usually 1) for
154 that level. By default locks are not shared.
156 This function can also define a list of tasklets, which then will be
157 executed in order instead of the usual LU-level CheckPrereq and Exec
158 functions, if those are not defined by the LU.
162 # Acquire all nodes and one instance
163 self.needed_locks = {
164 locking.LEVEL_NODE: locking.ALL_SET,
165 locking.LEVEL_INSTANCE: ['instance1.example.tld'],
167 # Acquire just two nodes
168 self.needed_locks = {
169 locking.LEVEL_NODE: ['node1.example.tld', 'node2.example.tld'],
172 self.needed_locks = {} # No, you can't leave it to the default value None
175 # The implementation of this method is mandatory only if the new LU is
176 # concurrent, so that old LUs don't need to be changed all at the same
179 self.needed_locks = {} # Exclusive LUs don't need locks.
181 raise NotImplementedError
183 def DeclareLocks(self, level):
184 """Declare LU locking needs for a level
186 While most LUs can just declare their locking needs at ExpandNames time,
187 sometimes there's the need to calculate some locks after having acquired
188 the ones before. This function is called just before acquiring locks at a
189 particular level, but after acquiring the ones at lower levels, and permits
190 such calculations. It can be used to modify self.needed_locks, and by
191 default it does nothing.
193 This function is only called if you have something already set in
194 self.needed_locks for the level.
196 @param level: Locking level which is going to be locked
197 @type level: member of ganeti.locking.LEVELS
201 def CheckPrereq(self):
202 """Check prerequisites for this LU.
204 This method should check that the prerequisites for the execution
205 of this LU are fulfilled. It can do internode communication, but
206 it should be idempotent - no cluster or system changes are
209 The method should raise errors.OpPrereqError in case something is
210 not fulfilled. Its return value is ignored.
212 This method should also update all the parameters of the opcode to
213 their canonical form if it hasn't been done by ExpandNames before.
216 if self.tasklets is not None:
217 for (idx, tl) in enumerate(self.tasklets):
218 logging.debug("Checking prerequisites for tasklet %s/%s",
219 idx + 1, len(self.tasklets))
222 raise NotImplementedError
224 def Exec(self, feedback_fn):
227 This method should implement the actual work. It should raise
228 errors.OpExecError for failures that are somewhat dealt with in
232 if self.tasklets is not None:
233 for (idx, tl) in enumerate(self.tasklets):
234 logging.debug("Executing tasklet %s/%s", idx + 1, len(self.tasklets))
237 raise NotImplementedError
239 def BuildHooksEnv(self):
240 """Build hooks environment for this LU.
242 This method should return a three-node tuple consisting of: a dict
243 containing the environment that will be used for running the
244 specific hook for this LU, a list of node names on which the hook
245 should run before the execution, and a list of node names on which
246 the hook should run after the execution.
248 The keys of the dict must not have 'GANETI_' prefixed as this will
249 be handled in the hooks runner. Also note additional keys will be
250 added by the hooks runner. If the LU doesn't define any
251 environment, an empty dict (and not None) should be returned.
253 No nodes should be returned as an empty list (and not None).
255 Note that if the HPATH for a LU class is None, this function will
259 raise NotImplementedError
261 def HooksCallBack(self, phase, hook_results, feedback_fn, lu_result):
262 """Notify the LU about the results of its hooks.
264 This method is called every time a hooks phase is executed, and notifies
265 the Logical Unit about the hooks' result. The LU can then use it to alter
266 its result based on the hooks. By default the method does nothing and the
267 previous result is passed back unchanged but any LU can define it if it
268 wants to use the local cluster hook-scripts somehow.
270 @param phase: one of L{constants.HOOKS_PHASE_POST} or
271 L{constants.HOOKS_PHASE_PRE}; it denotes the hooks phase
272 @param hook_results: the results of the multi-node hooks rpc call
273 @param feedback_fn: function used send feedback back to the caller
274 @param lu_result: the previous Exec result this LU had, or None
276 @return: the new Exec result, based on the previous result
282 def _ExpandAndLockInstance(self):
283 """Helper function to expand and lock an instance.
285 Many LUs that work on an instance take its name in self.op.instance_name
286 and need to expand it and then declare the expanded name for locking. This
287 function does it, and then updates self.op.instance_name to the expanded
288 name. It also initializes needed_locks as a dict, if this hasn't been done
292 if self.needed_locks is None:
293 self.needed_locks = {}
295 assert locking.LEVEL_INSTANCE not in self.needed_locks, \
296 "_ExpandAndLockInstance called with instance-level locks set"
297 expanded_name = self.cfg.ExpandInstanceName(self.op.instance_name)
298 if expanded_name is None:
299 raise errors.OpPrereqError("Instance '%s' not known" %
300 self.op.instance_name)
301 self.needed_locks[locking.LEVEL_INSTANCE] = expanded_name
302 self.op.instance_name = expanded_name
304 def _LockInstancesNodes(self, primary_only=False):
305 """Helper function to declare instances' nodes for locking.
307 This function should be called after locking one or more instances to lock
308 their nodes. Its effect is populating self.needed_locks[locking.LEVEL_NODE]
309 with all primary or secondary nodes for instances already locked and
310 present in self.needed_locks[locking.LEVEL_INSTANCE].
312 It should be called from DeclareLocks, and for safety only works if
313 self.recalculate_locks[locking.LEVEL_NODE] is set.
315 In the future it may grow parameters to just lock some instance's nodes, or
316 to just lock primaries or secondary nodes, if needed.
318 If should be called in DeclareLocks in a way similar to::
320 if level == locking.LEVEL_NODE:
321 self._LockInstancesNodes()
323 @type primary_only: boolean
324 @param primary_only: only lock primary nodes of locked instances
327 assert locking.LEVEL_NODE in self.recalculate_locks, \
328 "_LockInstancesNodes helper function called with no nodes to recalculate"
330 # TODO: check if we're really been called with the instance locks held
332 # For now we'll replace self.needed_locks[locking.LEVEL_NODE], but in the
333 # future we might want to have different behaviors depending on the value
334 # of self.recalculate_locks[locking.LEVEL_NODE]
336 for instance_name in self.acquired_locks[locking.LEVEL_INSTANCE]:
337 instance = self.context.cfg.GetInstanceInfo(instance_name)
338 wanted_nodes.append(instance.primary_node)
340 wanted_nodes.extend(instance.secondary_nodes)
342 if self.recalculate_locks[locking.LEVEL_NODE] == constants.LOCKS_REPLACE:
343 self.needed_locks[locking.LEVEL_NODE] = wanted_nodes
344 elif self.recalculate_locks[locking.LEVEL_NODE] == constants.LOCKS_APPEND:
345 self.needed_locks[locking.LEVEL_NODE].extend(wanted_nodes)
347 del self.recalculate_locks[locking.LEVEL_NODE]
350 class NoHooksLU(LogicalUnit):
351 """Simple LU which runs no hooks.
353 This LU is intended as a parent for other LogicalUnits which will
354 run no hooks, in order to reduce duplicate code.
362 """Tasklet base class.
364 Tasklets are subcomponents for LUs. LUs can consist entirely of tasklets or
365 they can mix legacy code with tasklets. Locking needs to be done in the LU,
366 tasklets know nothing about locks.
368 Subclasses must follow these rules:
369 - Implement CheckPrereq
373 def __init__(self, lu):
380 def CheckPrereq(self):
381 """Check prerequisites for this tasklets.
383 This method should check whether the prerequisites for the execution of
384 this tasklet are fulfilled. It can do internode communication, but it
385 should be idempotent - no cluster or system changes are allowed.
387 The method should raise errors.OpPrereqError in case something is not
388 fulfilled. Its return value is ignored.
390 This method should also update all parameters to their canonical form if it
391 hasn't been done before.
394 raise NotImplementedError
396 def Exec(self, feedback_fn):
397 """Execute the tasklet.
399 This method should implement the actual work. It should raise
400 errors.OpExecError for failures that are somewhat dealt with in code, or
404 raise NotImplementedError
407 def _GetWantedNodes(lu, nodes):
408 """Returns list of checked and expanded node names.
410 @type lu: L{LogicalUnit}
411 @param lu: the logical unit on whose behalf we execute
413 @param nodes: list of node names or None for all nodes
415 @return: the list of nodes, sorted
416 @raise errors.OpProgrammerError: if the nodes parameter is wrong type
419 if not isinstance(nodes, list):
420 raise errors.OpPrereqError("Invalid argument type 'nodes'")
423 raise errors.ProgrammerError("_GetWantedNodes should only be called with a"
424 " non-empty list of nodes whose name is to be expanded.")
428 node = lu.cfg.ExpandNodeName(name)
430 raise errors.OpPrereqError("No such node name '%s'" % name)
433 return utils.NiceSort(wanted)
436 def _GetWantedInstances(lu, instances):
437 """Returns list of checked and expanded instance names.
439 @type lu: L{LogicalUnit}
440 @param lu: the logical unit on whose behalf we execute
441 @type instances: list
442 @param instances: list of instance names or None for all instances
444 @return: the list of instances, sorted
445 @raise errors.OpPrereqError: if the instances parameter is wrong type
446 @raise errors.OpPrereqError: if any of the passed instances is not found
449 if not isinstance(instances, list):
450 raise errors.OpPrereqError("Invalid argument type 'instances'")
455 for name in instances:
456 instance = lu.cfg.ExpandInstanceName(name)
458 raise errors.OpPrereqError("No such instance name '%s'" % name)
459 wanted.append(instance)
462 wanted = utils.NiceSort(lu.cfg.GetInstanceList())
466 def _CheckOutputFields(static, dynamic, selected):
467 """Checks whether all selected fields are valid.
469 @type static: L{utils.FieldSet}
470 @param static: static fields set
471 @type dynamic: L{utils.FieldSet}
472 @param dynamic: dynamic fields set
479 delta = f.NonMatching(selected)
481 raise errors.OpPrereqError("Unknown output fields selected: %s"
485 def _CheckBooleanOpField(op, name):
486 """Validates boolean opcode parameters.
488 This will ensure that an opcode parameter is either a boolean value,
489 or None (but that it always exists).
492 val = getattr(op, name, None)
493 if not (val is None or isinstance(val, bool)):
494 raise errors.OpPrereqError("Invalid boolean parameter '%s' (%s)" %
496 setattr(op, name, val)
499 def _CheckNodeOnline(lu, node):
500 """Ensure that a given node is online.
502 @param lu: the LU on behalf of which we make the check
503 @param node: the node to check
504 @raise errors.OpPrereqError: if the node is offline
507 if lu.cfg.GetNodeInfo(node).offline:
508 raise errors.OpPrereqError("Can't use offline node %s" % node)
511 def _CheckNodeNotDrained(lu, node):
512 """Ensure that a given node is not drained.
514 @param lu: the LU on behalf of which we make the check
515 @param node: the node to check
516 @raise errors.OpPrereqError: if the node is drained
519 if lu.cfg.GetNodeInfo(node).drained:
520 raise errors.OpPrereqError("Can't use drained node %s" % node)
523 def _BuildInstanceHookEnv(name, primary_node, secondary_nodes, os_type, status,
524 memory, vcpus, nics, disk_template, disks,
525 bep, hvp, hypervisor_name):
526 """Builds instance related env variables for hooks
528 This builds the hook environment from individual variables.
531 @param name: the name of the instance
532 @type primary_node: string
533 @param primary_node: the name of the instance's primary node
534 @type secondary_nodes: list
535 @param secondary_nodes: list of secondary nodes as strings
536 @type os_type: string
537 @param os_type: the name of the instance's OS
538 @type status: boolean
539 @param status: the should_run status of the instance
541 @param memory: the memory size of the instance
543 @param vcpus: the count of VCPUs the instance has
545 @param nics: list of tuples (ip, mac, mode, link) representing
546 the NICs the instance has
547 @type disk_template: string
548 @param disk_template: the disk template of the instance
550 @param disks: the list of (size, mode) pairs
552 @param bep: the backend parameters for the instance
554 @param hvp: the hypervisor parameters for the instance
555 @type hypervisor_name: string
556 @param hypervisor_name: the hypervisor for the instance
558 @return: the hook environment for this instance
567 "INSTANCE_NAME": name,
568 "INSTANCE_PRIMARY": primary_node,
569 "INSTANCE_SECONDARIES": " ".join(secondary_nodes),
570 "INSTANCE_OS_TYPE": os_type,
571 "INSTANCE_STATUS": str_status,
572 "INSTANCE_MEMORY": memory,
573 "INSTANCE_VCPUS": vcpus,
574 "INSTANCE_DISK_TEMPLATE": disk_template,
575 "INSTANCE_HYPERVISOR": hypervisor_name,
579 nic_count = len(nics)
580 for idx, (ip, mac, mode, link) in enumerate(nics):
583 env["INSTANCE_NIC%d_IP" % idx] = ip
584 env["INSTANCE_NIC%d_MAC" % idx] = mac
585 env["INSTANCE_NIC%d_MODE" % idx] = mode
586 env["INSTANCE_NIC%d_LINK" % idx] = link
587 if mode == constants.NIC_MODE_BRIDGED:
588 env["INSTANCE_NIC%d_BRIDGE" % idx] = link
592 env["INSTANCE_NIC_COUNT"] = nic_count
595 disk_count = len(disks)
596 for idx, (size, mode) in enumerate(disks):
597 env["INSTANCE_DISK%d_SIZE" % idx] = size
598 env["INSTANCE_DISK%d_MODE" % idx] = mode
602 env["INSTANCE_DISK_COUNT"] = disk_count
604 for source, kind in [(bep, "BE"), (hvp, "HV")]:
605 for key, value in source.items():
606 env["INSTANCE_%s_%s" % (kind, key)] = value
611 def _NICListToTuple(lu, nics):
612 """Build a list of nic information tuples.
614 This list is suitable to be passed to _BuildInstanceHookEnv or as a return
615 value in LUQueryInstanceData.
617 @type lu: L{LogicalUnit}
618 @param lu: the logical unit on whose behalf we execute
619 @type nics: list of L{objects.NIC}
620 @param nics: list of nics to convert to hooks tuples
624 c_nicparams = lu.cfg.GetClusterInfo().nicparams[constants.PP_DEFAULT]
628 filled_params = objects.FillDict(c_nicparams, nic.nicparams)
629 mode = filled_params[constants.NIC_MODE]
630 link = filled_params[constants.NIC_LINK]
631 hooks_nics.append((ip, mac, mode, link))
635 def _BuildInstanceHookEnvByObject(lu, instance, override=None):
636 """Builds instance related env variables for hooks from an object.
638 @type lu: L{LogicalUnit}
639 @param lu: the logical unit on whose behalf we execute
640 @type instance: L{objects.Instance}
641 @param instance: the instance for which we should build the
644 @param override: dictionary with key/values that will override
647 @return: the hook environment dictionary
650 cluster = lu.cfg.GetClusterInfo()
651 bep = cluster.FillBE(instance)
652 hvp = cluster.FillHV(instance)
654 'name': instance.name,
655 'primary_node': instance.primary_node,
656 'secondary_nodes': instance.secondary_nodes,
657 'os_type': instance.os,
658 'status': instance.admin_up,
659 'memory': bep[constants.BE_MEMORY],
660 'vcpus': bep[constants.BE_VCPUS],
661 'nics': _NICListToTuple(lu, instance.nics),
662 'disk_template': instance.disk_template,
663 'disks': [(disk.size, disk.mode) for disk in instance.disks],
666 'hypervisor_name': instance.hypervisor,
669 args.update(override)
670 return _BuildInstanceHookEnv(**args)
673 def _AdjustCandidatePool(lu):
674 """Adjust the candidate pool after node operations.
677 mod_list = lu.cfg.MaintainCandidatePool()
679 lu.LogInfo("Promoted nodes to master candidate role: %s",
680 ", ".join(node.name for node in mod_list))
681 for name in mod_list:
682 lu.context.ReaddNode(name)
683 mc_now, mc_max = lu.cfg.GetMasterCandidateStats()
685 lu.LogInfo("Note: more nodes are candidates (%d) than desired (%d)" %
689 def _CheckNicsBridgesExist(lu, target_nics, target_node,
690 profile=constants.PP_DEFAULT):
691 """Check that the brigdes needed by a list of nics exist.
694 c_nicparams = lu.cfg.GetClusterInfo().nicparams[profile]
695 paramslist = [objects.FillDict(c_nicparams, nic.nicparams)
696 for nic in target_nics]
697 brlist = [params[constants.NIC_LINK] for params in paramslist
698 if params[constants.NIC_MODE] == constants.NIC_MODE_BRIDGED]
700 result = lu.rpc.call_bridges_exist(target_node, brlist)
701 result.Raise("Error checking bridges on destination node '%s'" %
702 target_node, prereq=True)
705 def _CheckInstanceBridgesExist(lu, instance, node=None):
706 """Check that the brigdes needed by an instance exist.
710 node = instance.primary_node
711 _CheckNicsBridgesExist(lu, instance.nics, node)
714 def _GetNodePrimaryInstances(cfg, node_name):
715 """Returns primary instances on a node.
720 for (_, inst) in cfg.GetAllInstancesInfo().iteritems():
721 if node_name == inst.primary_node:
722 instances.append(inst)
727 def _GetNodeSecondaryInstances(cfg, node_name):
728 """Returns secondary instances on a node.
733 for (_, inst) in cfg.GetAllInstancesInfo().iteritems():
734 if node_name in inst.secondary_nodes:
735 instances.append(inst)
740 def _GetStorageTypeArgs(cfg, storage_type):
741 """Returns the arguments for a storage type.
744 # Special case for file storage
745 if storage_type == constants.ST_FILE:
746 # storage.FileStorage wants a list of storage directories
747 return [[cfg.GetFileStorageDir()]]
752 class LUPostInitCluster(LogicalUnit):
753 """Logical unit for running hooks after cluster initialization.
756 HPATH = "cluster-init"
757 HTYPE = constants.HTYPE_CLUSTER
760 def BuildHooksEnv(self):
764 env = {"OP_TARGET": self.cfg.GetClusterName()}
765 mn = self.cfg.GetMasterNode()
768 def CheckPrereq(self):
769 """No prerequisites to check.
774 def Exec(self, feedback_fn):
781 class LUDestroyCluster(NoHooksLU):
782 """Logical unit for destroying the cluster.
787 def CheckPrereq(self):
788 """Check prerequisites.
790 This checks whether the cluster is empty.
792 Any errors are signaled by raising errors.OpPrereqError.
795 master = self.cfg.GetMasterNode()
797 nodelist = self.cfg.GetNodeList()
798 if len(nodelist) != 1 or nodelist[0] != master:
799 raise errors.OpPrereqError("There are still %d node(s) in"
800 " this cluster." % (len(nodelist) - 1))
801 instancelist = self.cfg.GetInstanceList()
803 raise errors.OpPrereqError("There are still %d instance(s) in"
804 " this cluster." % len(instancelist))
806 def Exec(self, feedback_fn):
807 """Destroys the cluster.
810 master = self.cfg.GetMasterNode()
811 result = self.rpc.call_node_stop_master(master, False)
812 result.Raise("Could not disable the master role")
813 priv_key, pub_key, _ = ssh.GetUserFiles(constants.GANETI_RUNAS)
814 utils.CreateBackup(priv_key)
815 utils.CreateBackup(pub_key)
819 class LUVerifyCluster(LogicalUnit):
820 """Verifies the cluster status.
823 HPATH = "cluster-verify"
824 HTYPE = constants.HTYPE_CLUSTER
825 _OP_REQP = ["skip_checks"]
828 def ExpandNames(self):
829 self.needed_locks = {
830 locking.LEVEL_NODE: locking.ALL_SET,
831 locking.LEVEL_INSTANCE: locking.ALL_SET,
833 self.share_locks = dict.fromkeys(locking.LEVELS, 1)
835 def _VerifyNode(self, nodeinfo, file_list, local_cksum,
836 node_result, feedback_fn, master_files,
838 """Run multiple tests against a node.
842 - compares ganeti version
843 - checks vg existence and size > 20G
844 - checks config file checksum
845 - checks ssh to other nodes
847 @type nodeinfo: L{objects.Node}
848 @param nodeinfo: the node to check
849 @param file_list: required list of files
850 @param local_cksum: dictionary of local files and their checksums
851 @param node_result: the results from the node
852 @param feedback_fn: function used to accumulate results
853 @param master_files: list of files that only masters should have
854 @param drbd_map: the useddrbd minors for this node, in
855 form of minor: (instance, must_exist) which correspond to instances
856 and their running status
857 @param vg_name: Ganeti Volume Group (result of self.cfg.GetVGName())
862 # main result, node_result should be a non-empty dict
863 if not node_result or not isinstance(node_result, dict):
864 feedback_fn(" - ERROR: unable to verify node %s." % (node,))
867 # compares ganeti version
868 local_version = constants.PROTOCOL_VERSION
869 remote_version = node_result.get('version', None)
870 if not (remote_version and isinstance(remote_version, (list, tuple)) and
871 len(remote_version) == 2):
872 feedback_fn(" - ERROR: connection to %s failed" % (node))
875 if local_version != remote_version[0]:
876 feedback_fn(" - ERROR: incompatible protocol versions: master %s,"
877 " node %s %s" % (local_version, node, remote_version[0]))
880 # node seems compatible, we can actually try to look into its results
884 # full package version
885 if constants.RELEASE_VERSION != remote_version[1]:
886 feedback_fn(" - WARNING: software version mismatch: master %s,"
888 (constants.RELEASE_VERSION, node, remote_version[1]))
890 # checks vg existence and size > 20G
891 if vg_name is not None:
892 vglist = node_result.get(constants.NV_VGLIST, None)
894 feedback_fn(" - ERROR: unable to check volume groups on node %s." %
898 vgstatus = utils.CheckVolumeGroupSize(vglist, vg_name,
899 constants.MIN_VG_SIZE)
901 feedback_fn(" - ERROR: %s on node %s" % (vgstatus, node))
904 # checks config file checksum
906 remote_cksum = node_result.get(constants.NV_FILELIST, None)
907 if not isinstance(remote_cksum, dict):
909 feedback_fn(" - ERROR: node hasn't returned file checksum data")
911 for file_name in file_list:
912 node_is_mc = nodeinfo.master_candidate
913 must_have_file = file_name not in master_files
914 if file_name not in remote_cksum:
915 if node_is_mc or must_have_file:
917 feedback_fn(" - ERROR: file '%s' missing" % file_name)
918 elif remote_cksum[file_name] != local_cksum[file_name]:
919 if node_is_mc or must_have_file:
921 feedback_fn(" - ERROR: file '%s' has wrong checksum" % file_name)
923 # not candidate and this is not a must-have file
925 feedback_fn(" - ERROR: file '%s' should not exist on non master"
926 " candidates (and the file is outdated)" % file_name)
928 # all good, except non-master/non-must have combination
929 if not node_is_mc and not must_have_file:
930 feedback_fn(" - ERROR: file '%s' should not exist on non master"
931 " candidates" % file_name)
935 if constants.NV_NODELIST not in node_result:
937 feedback_fn(" - ERROR: node hasn't returned node ssh connectivity data")
939 if node_result[constants.NV_NODELIST]:
941 for node in node_result[constants.NV_NODELIST]:
942 feedback_fn(" - ERROR: ssh communication with node '%s': %s" %
943 (node, node_result[constants.NV_NODELIST][node]))
945 if constants.NV_NODENETTEST not in node_result:
947 feedback_fn(" - ERROR: node hasn't returned node tcp connectivity data")
949 if node_result[constants.NV_NODENETTEST]:
951 nlist = utils.NiceSort(node_result[constants.NV_NODENETTEST].keys())
953 feedback_fn(" - ERROR: tcp communication with node '%s': %s" %
954 (node, node_result[constants.NV_NODENETTEST][node]))
956 hyp_result = node_result.get(constants.NV_HYPERVISOR, None)
957 if isinstance(hyp_result, dict):
958 for hv_name, hv_result in hyp_result.iteritems():
959 if hv_result is not None:
960 feedback_fn(" - ERROR: hypervisor %s verify failure: '%s'" %
961 (hv_name, hv_result))
963 # check used drbd list
964 if vg_name is not None:
965 used_minors = node_result.get(constants.NV_DRBDLIST, [])
966 if not isinstance(used_minors, (tuple, list)):
967 feedback_fn(" - ERROR: cannot parse drbd status file: %s" %
970 for minor, (iname, must_exist) in drbd_map.items():
971 if minor not in used_minors and must_exist:
972 feedback_fn(" - ERROR: drbd minor %d of instance %s is"
973 " not active" % (minor, iname))
975 for minor in used_minors:
976 if minor not in drbd_map:
977 feedback_fn(" - ERROR: unallocated drbd minor %d is in use" %
983 def _VerifyInstance(self, instance, instanceconfig, node_vol_is,
984 node_instance, feedback_fn, n_offline):
985 """Verify an instance.
987 This function checks to see if the required block devices are
988 available on the instance's node.
993 node_current = instanceconfig.primary_node
996 instanceconfig.MapLVsByNode(node_vol_should)
998 for node in node_vol_should:
999 if node in n_offline:
1000 # ignore missing volumes on offline nodes
1002 for volume in node_vol_should[node]:
1003 if node not in node_vol_is or volume not in node_vol_is[node]:
1004 feedback_fn(" - ERROR: volume %s missing on node %s" %
1008 if instanceconfig.admin_up:
1009 if ((node_current not in node_instance or
1010 not instance in node_instance[node_current]) and
1011 node_current not in n_offline):
1012 feedback_fn(" - ERROR: instance %s not running on node %s" %
1013 (instance, node_current))
1016 for node in node_instance:
1017 if (not node == node_current):
1018 if instance in node_instance[node]:
1019 feedback_fn(" - ERROR: instance %s should not run on node %s" %
1025 def _VerifyOrphanVolumes(self, node_vol_should, node_vol_is, feedback_fn):
1026 """Verify if there are any unknown volumes in the cluster.
1028 The .os, .swap and backup volumes are ignored. All other volumes are
1029 reported as unknown.
1034 for node in node_vol_is:
1035 for volume in node_vol_is[node]:
1036 if node not in node_vol_should or volume not in node_vol_should[node]:
1037 feedback_fn(" - ERROR: volume %s on node %s should not exist" %
1042 def _VerifyOrphanInstances(self, instancelist, node_instance, feedback_fn):
1043 """Verify the list of running instances.
1045 This checks what instances are running but unknown to the cluster.
1049 for node in node_instance:
1050 for runninginstance in node_instance[node]:
1051 if runninginstance not in instancelist:
1052 feedback_fn(" - ERROR: instance %s on node %s should not exist" %
1053 (runninginstance, node))
1057 def _VerifyNPlusOneMemory(self, node_info, instance_cfg, feedback_fn):
1058 """Verify N+1 Memory Resilience.
1060 Check that if one single node dies we can still start all the instances it
1066 for node, nodeinfo in node_info.iteritems():
1067 # This code checks that every node which is now listed as secondary has
1068 # enough memory to host all instances it is supposed to should a single
1069 # other node in the cluster fail.
1070 # FIXME: not ready for failover to an arbitrary node
1071 # FIXME: does not support file-backed instances
1072 # WARNING: we currently take into account down instances as well as up
1073 # ones, considering that even if they're down someone might want to start
1074 # them even in the event of a node failure.
1075 for prinode, instances in nodeinfo['sinst-by-pnode'].iteritems():
1077 for instance in instances:
1078 bep = self.cfg.GetClusterInfo().FillBE(instance_cfg[instance])
1079 if bep[constants.BE_AUTO_BALANCE]:
1080 needed_mem += bep[constants.BE_MEMORY]
1081 if nodeinfo['mfree'] < needed_mem:
1082 feedback_fn(" - ERROR: not enough memory on node %s to accommodate"
1083 " failovers should node %s fail" % (node, prinode))
1087 def CheckPrereq(self):
1088 """Check prerequisites.
1090 Transform the list of checks we're going to skip into a set and check that
1091 all its members are valid.
1094 self.skip_set = frozenset(self.op.skip_checks)
1095 if not constants.VERIFY_OPTIONAL_CHECKS.issuperset(self.skip_set):
1096 raise errors.OpPrereqError("Invalid checks to be skipped specified")
1098 def BuildHooksEnv(self):
1101 Cluster-Verify hooks just ran in the post phase and their failure makes
1102 the output be logged in the verify output and the verification to fail.
1105 all_nodes = self.cfg.GetNodeList()
1107 "CLUSTER_TAGS": " ".join(self.cfg.GetClusterInfo().GetTags())
1109 for node in self.cfg.GetAllNodesInfo().values():
1110 env["NODE_TAGS_%s" % node.name] = " ".join(node.GetTags())
1112 return env, [], all_nodes
1114 def Exec(self, feedback_fn):
1115 """Verify integrity of cluster, performing various test on nodes.
1119 feedback_fn("* Verifying global settings")
1120 for msg in self.cfg.VerifyConfig():
1121 feedback_fn(" - ERROR: %s" % msg)
1123 vg_name = self.cfg.GetVGName()
1124 hypervisors = self.cfg.GetClusterInfo().enabled_hypervisors
1125 nodelist = utils.NiceSort(self.cfg.GetNodeList())
1126 nodeinfo = [self.cfg.GetNodeInfo(nname) for nname in nodelist]
1127 instancelist = utils.NiceSort(self.cfg.GetInstanceList())
1128 instanceinfo = dict((iname, self.cfg.GetInstanceInfo(iname))
1129 for iname in instancelist)
1130 i_non_redundant = [] # Non redundant instances
1131 i_non_a_balanced = [] # Non auto-balanced instances
1132 n_offline = [] # List of offline nodes
1133 n_drained = [] # List of nodes being drained
1139 # FIXME: verify OS list
1140 # do local checksums
1141 master_files = [constants.CLUSTER_CONF_FILE]
1143 file_names = ssconf.SimpleStore().GetFileList()
1144 file_names.append(constants.SSL_CERT_FILE)
1145 file_names.append(constants.RAPI_CERT_FILE)
1146 file_names.extend(master_files)
1148 local_checksums = utils.FingerprintFiles(file_names)
1150 feedback_fn("* Gathering data (%d nodes)" % len(nodelist))
1151 node_verify_param = {
1152 constants.NV_FILELIST: file_names,
1153 constants.NV_NODELIST: [node.name for node in nodeinfo
1154 if not node.offline],
1155 constants.NV_HYPERVISOR: hypervisors,
1156 constants.NV_NODENETTEST: [(node.name, node.primary_ip,
1157 node.secondary_ip) for node in nodeinfo
1158 if not node.offline],
1159 constants.NV_INSTANCELIST: hypervisors,
1160 constants.NV_VERSION: None,
1161 constants.NV_HVINFO: self.cfg.GetHypervisorType(),
1163 if vg_name is not None:
1164 node_verify_param[constants.NV_VGLIST] = None
1165 node_verify_param[constants.NV_LVLIST] = vg_name
1166 node_verify_param[constants.NV_DRBDLIST] = None
1167 all_nvinfo = self.rpc.call_node_verify(nodelist, node_verify_param,
1168 self.cfg.GetClusterName())
1170 cluster = self.cfg.GetClusterInfo()
1171 master_node = self.cfg.GetMasterNode()
1172 all_drbd_map = self.cfg.ComputeDRBDMap()
1174 for node_i in nodeinfo:
1178 feedback_fn("* Skipping offline node %s" % (node,))
1179 n_offline.append(node)
1182 if node == master_node:
1184 elif node_i.master_candidate:
1185 ntype = "master candidate"
1186 elif node_i.drained:
1188 n_drained.append(node)
1191 feedback_fn("* Verifying node %s (%s)" % (node, ntype))
1193 msg = all_nvinfo[node].fail_msg
1195 feedback_fn(" - ERROR: while contacting node %s: %s" % (node, msg))
1199 nresult = all_nvinfo[node].payload
1201 for minor, instance in all_drbd_map[node].items():
1202 if instance not in instanceinfo:
1203 feedback_fn(" - ERROR: ghost instance '%s' in temporary DRBD map" %
1205 # ghost instance should not be running, but otherwise we
1206 # don't give double warnings (both ghost instance and
1207 # unallocated minor in use)
1208 node_drbd[minor] = (instance, False)
1210 instance = instanceinfo[instance]
1211 node_drbd[minor] = (instance.name, instance.admin_up)
1212 result = self._VerifyNode(node_i, file_names, local_checksums,
1213 nresult, feedback_fn, master_files,
1217 lvdata = nresult.get(constants.NV_LVLIST, "Missing LV data")
1219 node_volume[node] = {}
1220 elif isinstance(lvdata, basestring):
1221 feedback_fn(" - ERROR: LVM problem on node %s: %s" %
1222 (node, utils.SafeEncode(lvdata)))
1224 node_volume[node] = {}
1225 elif not isinstance(lvdata, dict):
1226 feedback_fn(" - ERROR: connection to %s failed (lvlist)" % (node,))
1230 node_volume[node] = lvdata
1233 idata = nresult.get(constants.NV_INSTANCELIST, None)
1234 if not isinstance(idata, list):
1235 feedback_fn(" - ERROR: connection to %s failed (instancelist)" %
1240 node_instance[node] = idata
1243 nodeinfo = nresult.get(constants.NV_HVINFO, None)
1244 if not isinstance(nodeinfo, dict):
1245 feedback_fn(" - ERROR: connection to %s failed (hvinfo)" % (node,))
1251 "mfree": int(nodeinfo['memory_free']),
1254 # dictionary holding all instances this node is secondary for,
1255 # grouped by their primary node. Each key is a cluster node, and each
1256 # value is a list of instances which have the key as primary and the
1257 # current node as secondary. this is handy to calculate N+1 memory
1258 # availability if you can only failover from a primary to its
1260 "sinst-by-pnode": {},
1262 # FIXME: devise a free space model for file based instances as well
1263 if vg_name is not None:
1264 if (constants.NV_VGLIST not in nresult or
1265 vg_name not in nresult[constants.NV_VGLIST]):
1266 feedback_fn(" - ERROR: node %s didn't return data for the"
1267 " volume group '%s' - it is either missing or broken" %
1271 node_info[node]["dfree"] = int(nresult[constants.NV_VGLIST][vg_name])
1272 except (ValueError, KeyError):
1273 feedback_fn(" - ERROR: invalid nodeinfo value returned"
1274 " from node %s" % (node,))
1278 node_vol_should = {}
1280 for instance in instancelist:
1281 feedback_fn("* Verifying instance %s" % instance)
1282 inst_config = instanceinfo[instance]
1283 result = self._VerifyInstance(instance, inst_config, node_volume,
1284 node_instance, feedback_fn, n_offline)
1286 inst_nodes_offline = []
1288 inst_config.MapLVsByNode(node_vol_should)
1290 instance_cfg[instance] = inst_config
1292 pnode = inst_config.primary_node
1293 if pnode in node_info:
1294 node_info[pnode]['pinst'].append(instance)
1295 elif pnode not in n_offline:
1296 feedback_fn(" - ERROR: instance %s, connection to primary node"
1297 " %s failed" % (instance, pnode))
1300 if pnode in n_offline:
1301 inst_nodes_offline.append(pnode)
1303 # If the instance is non-redundant we cannot survive losing its primary
1304 # node, so we are not N+1 compliant. On the other hand we have no disk
1305 # templates with more than one secondary so that situation is not well
1307 # FIXME: does not support file-backed instances
1308 if len(inst_config.secondary_nodes) == 0:
1309 i_non_redundant.append(instance)
1310 elif len(inst_config.secondary_nodes) > 1:
1311 feedback_fn(" - WARNING: multiple secondaries for instance %s"
1314 if not cluster.FillBE(inst_config)[constants.BE_AUTO_BALANCE]:
1315 i_non_a_balanced.append(instance)
1317 for snode in inst_config.secondary_nodes:
1318 if snode in node_info:
1319 node_info[snode]['sinst'].append(instance)
1320 if pnode not in node_info[snode]['sinst-by-pnode']:
1321 node_info[snode]['sinst-by-pnode'][pnode] = []
1322 node_info[snode]['sinst-by-pnode'][pnode].append(instance)
1323 elif snode not in n_offline:
1324 feedback_fn(" - ERROR: instance %s, connection to secondary node"
1325 " %s failed" % (instance, snode))
1327 if snode in n_offline:
1328 inst_nodes_offline.append(snode)
1330 if inst_nodes_offline:
1331 # warn that the instance lives on offline nodes, and set bad=True
1332 feedback_fn(" - ERROR: instance lives on offline node(s) %s" %
1333 ", ".join(inst_nodes_offline))
1336 feedback_fn("* Verifying orphan volumes")
1337 result = self._VerifyOrphanVolumes(node_vol_should, node_volume,
1341 feedback_fn("* Verifying remaining instances")
1342 result = self._VerifyOrphanInstances(instancelist, node_instance,
1346 if constants.VERIFY_NPLUSONE_MEM not in self.skip_set:
1347 feedback_fn("* Verifying N+1 Memory redundancy")
1348 result = self._VerifyNPlusOneMemory(node_info, instance_cfg, feedback_fn)
1351 feedback_fn("* Other Notes")
1353 feedback_fn(" - NOTICE: %d non-redundant instance(s) found."
1354 % len(i_non_redundant))
1356 if i_non_a_balanced:
1357 feedback_fn(" - NOTICE: %d non-auto-balanced instance(s) found."
1358 % len(i_non_a_balanced))
1361 feedback_fn(" - NOTICE: %d offline node(s) found." % len(n_offline))
1364 feedback_fn(" - NOTICE: %d drained node(s) found." % len(n_drained))
1368 def HooksCallBack(self, phase, hooks_results, feedback_fn, lu_result):
1369 """Analyze the post-hooks' result
1371 This method analyses the hook result, handles it, and sends some
1372 nicely-formatted feedback back to the user.
1374 @param phase: one of L{constants.HOOKS_PHASE_POST} or
1375 L{constants.HOOKS_PHASE_PRE}; it denotes the hooks phase
1376 @param hooks_results: the results of the multi-node hooks rpc call
1377 @param feedback_fn: function used send feedback back to the caller
1378 @param lu_result: previous Exec result
1379 @return: the new Exec result, based on the previous result
1383 # We only really run POST phase hooks, and are only interested in
1385 if phase == constants.HOOKS_PHASE_POST:
1386 # Used to change hooks' output to proper indentation
1387 indent_re = re.compile('^', re.M)
1388 feedback_fn("* Hooks Results")
1389 if not hooks_results:
1390 feedback_fn(" - ERROR: general communication failure")
1393 for node_name in hooks_results:
1394 show_node_header = True
1395 res = hooks_results[node_name]
1399 # no need to warn or set fail return value
1401 feedback_fn(" Communication failure in hooks execution: %s" %
1405 for script, hkr, output in res.payload:
1406 if hkr == constants.HKR_FAIL:
1407 # The node header is only shown once, if there are
1408 # failing hooks on that node
1409 if show_node_header:
1410 feedback_fn(" Node %s:" % node_name)
1411 show_node_header = False
1412 feedback_fn(" ERROR: Script %s failed, output:" % script)
1413 output = indent_re.sub(' ', output)
1414 feedback_fn("%s" % output)
1420 class LUVerifyDisks(NoHooksLU):
1421 """Verifies the cluster disks status.
1427 def ExpandNames(self):
1428 self.needed_locks = {
1429 locking.LEVEL_NODE: locking.ALL_SET,
1430 locking.LEVEL_INSTANCE: locking.ALL_SET,
1432 self.share_locks = dict.fromkeys(locking.LEVELS, 1)
1434 def CheckPrereq(self):
1435 """Check prerequisites.
1437 This has no prerequisites.
1442 def Exec(self, feedback_fn):
1443 """Verify integrity of cluster disks.
1445 @rtype: tuple of three items
1446 @return: a tuple of (dict of node-to-node_error, list of instances
1447 which need activate-disks, dict of instance: (node, volume) for
1451 result = res_nodes, res_instances, res_missing = {}, [], {}
1453 vg_name = self.cfg.GetVGName()
1454 nodes = utils.NiceSort(self.cfg.GetNodeList())
1455 instances = [self.cfg.GetInstanceInfo(name)
1456 for name in self.cfg.GetInstanceList()]
1459 for inst in instances:
1461 if (not inst.admin_up or
1462 inst.disk_template not in constants.DTS_NET_MIRROR):
1464 inst.MapLVsByNode(inst_lvs)
1465 # transform { iname: {node: [vol,],},} to {(node, vol): iname}
1466 for node, vol_list in inst_lvs.iteritems():
1467 for vol in vol_list:
1468 nv_dict[(node, vol)] = inst
1473 node_lvs = self.rpc.call_lv_list(nodes, vg_name)
1477 node_res = node_lvs[node]
1478 if node_res.offline:
1480 msg = node_res.fail_msg
1482 logging.warning("Error enumerating LVs on node %s: %s", node, msg)
1483 res_nodes[node] = msg
1486 lvs = node_res.payload
1487 for lv_name, (_, lv_inactive, lv_online) in lvs.items():
1488 inst = nv_dict.pop((node, lv_name), None)
1489 if (not lv_online and inst is not None
1490 and inst.name not in res_instances):
1491 res_instances.append(inst.name)
1493 # any leftover items in nv_dict are missing LVs, let's arrange the
1495 for key, inst in nv_dict.iteritems():
1496 if inst.name not in res_missing:
1497 res_missing[inst.name] = []
1498 res_missing[inst.name].append(key)
1503 class LURenameCluster(LogicalUnit):
1504 """Rename the cluster.
1507 HPATH = "cluster-rename"
1508 HTYPE = constants.HTYPE_CLUSTER
1511 def BuildHooksEnv(self):
1516 "OP_TARGET": self.cfg.GetClusterName(),
1517 "NEW_NAME": self.op.name,
1519 mn = self.cfg.GetMasterNode()
1520 return env, [mn], [mn]
1522 def CheckPrereq(self):
1523 """Verify that the passed name is a valid one.
1526 hostname = utils.HostInfo(self.op.name)
1528 new_name = hostname.name
1529 self.ip = new_ip = hostname.ip
1530 old_name = self.cfg.GetClusterName()
1531 old_ip = self.cfg.GetMasterIP()
1532 if new_name == old_name and new_ip == old_ip:
1533 raise errors.OpPrereqError("Neither the name nor the IP address of the"
1534 " cluster has changed")
1535 if new_ip != old_ip:
1536 if utils.TcpPing(new_ip, constants.DEFAULT_NODED_PORT):
1537 raise errors.OpPrereqError("The given cluster IP address (%s) is"
1538 " reachable on the network. Aborting." %
1541 self.op.name = new_name
1543 def Exec(self, feedback_fn):
1544 """Rename the cluster.
1547 clustername = self.op.name
1550 # shutdown the master IP
1551 master = self.cfg.GetMasterNode()
1552 result = self.rpc.call_node_stop_master(master, False)
1553 result.Raise("Could not disable the master role")
1556 cluster = self.cfg.GetClusterInfo()
1557 cluster.cluster_name = clustername
1558 cluster.master_ip = ip
1559 self.cfg.Update(cluster)
1561 # update the known hosts file
1562 ssh.WriteKnownHostsFile(self.cfg, constants.SSH_KNOWN_HOSTS_FILE)
1563 node_list = self.cfg.GetNodeList()
1565 node_list.remove(master)
1568 result = self.rpc.call_upload_file(node_list,
1569 constants.SSH_KNOWN_HOSTS_FILE)
1570 for to_node, to_result in result.iteritems():
1571 msg = to_result.fail_msg
1573 msg = ("Copy of file %s to node %s failed: %s" %
1574 (constants.SSH_KNOWN_HOSTS_FILE, to_node, msg))
1575 self.proc.LogWarning(msg)
1578 result = self.rpc.call_node_start_master(master, False, False)
1579 msg = result.fail_msg
1581 self.LogWarning("Could not re-enable the master role on"
1582 " the master, please restart manually: %s", msg)
1585 def _RecursiveCheckIfLVMBased(disk):
1586 """Check if the given disk or its children are lvm-based.
1588 @type disk: L{objects.Disk}
1589 @param disk: the disk to check
1591 @return: boolean indicating whether a LD_LV dev_type was found or not
1595 for chdisk in disk.children:
1596 if _RecursiveCheckIfLVMBased(chdisk):
1598 return disk.dev_type == constants.LD_LV
1601 class LUSetClusterParams(LogicalUnit):
1602 """Change the parameters of the cluster.
1605 HPATH = "cluster-modify"
1606 HTYPE = constants.HTYPE_CLUSTER
1610 def CheckArguments(self):
1614 if not hasattr(self.op, "candidate_pool_size"):
1615 self.op.candidate_pool_size = None
1616 if self.op.candidate_pool_size is not None:
1618 self.op.candidate_pool_size = int(self.op.candidate_pool_size)
1619 except (ValueError, TypeError), err:
1620 raise errors.OpPrereqError("Invalid candidate_pool_size value: %s" %
1622 if self.op.candidate_pool_size < 1:
1623 raise errors.OpPrereqError("At least one master candidate needed")
1625 def ExpandNames(self):
1626 # FIXME: in the future maybe other cluster params won't require checking on
1627 # all nodes to be modified.
1628 self.needed_locks = {
1629 locking.LEVEL_NODE: locking.ALL_SET,
1631 self.share_locks[locking.LEVEL_NODE] = 1
1633 def BuildHooksEnv(self):
1638 "OP_TARGET": self.cfg.GetClusterName(),
1639 "NEW_VG_NAME": self.op.vg_name,
1641 mn = self.cfg.GetMasterNode()
1642 return env, [mn], [mn]
1644 def CheckPrereq(self):
1645 """Check prerequisites.
1647 This checks whether the given params don't conflict and
1648 if the given volume group is valid.
1651 if self.op.vg_name is not None and not self.op.vg_name:
1652 instances = self.cfg.GetAllInstancesInfo().values()
1653 for inst in instances:
1654 for disk in inst.disks:
1655 if _RecursiveCheckIfLVMBased(disk):
1656 raise errors.OpPrereqError("Cannot disable lvm storage while"
1657 " lvm-based instances exist")
1659 node_list = self.acquired_locks[locking.LEVEL_NODE]
1661 # if vg_name not None, checks given volume group on all nodes
1663 vglist = self.rpc.call_vg_list(node_list)
1664 for node in node_list:
1665 msg = vglist[node].fail_msg
1667 # ignoring down node
1668 self.LogWarning("Error while gathering data on node %s"
1669 " (ignoring node): %s", node, msg)
1671 vgstatus = utils.CheckVolumeGroupSize(vglist[node].payload,
1673 constants.MIN_VG_SIZE)
1675 raise errors.OpPrereqError("Error on node '%s': %s" %
1678 self.cluster = cluster = self.cfg.GetClusterInfo()
1679 # validate params changes
1680 if self.op.beparams:
1681 utils.ForceDictType(self.op.beparams, constants.BES_PARAMETER_TYPES)
1682 self.new_beparams = objects.FillDict(
1683 cluster.beparams[constants.PP_DEFAULT], self.op.beparams)
1685 if self.op.nicparams:
1686 utils.ForceDictType(self.op.nicparams, constants.NICS_PARAMETER_TYPES)
1687 self.new_nicparams = objects.FillDict(
1688 cluster.nicparams[constants.PP_DEFAULT], self.op.nicparams)
1689 objects.NIC.CheckParameterSyntax(self.new_nicparams)
1691 # hypervisor list/parameters
1692 self.new_hvparams = objects.FillDict(cluster.hvparams, {})
1693 if self.op.hvparams:
1694 if not isinstance(self.op.hvparams, dict):
1695 raise errors.OpPrereqError("Invalid 'hvparams' parameter on input")
1696 for hv_name, hv_dict in self.op.hvparams.items():
1697 if hv_name not in self.new_hvparams:
1698 self.new_hvparams[hv_name] = hv_dict
1700 self.new_hvparams[hv_name].update(hv_dict)
1702 if self.op.enabled_hypervisors is not None:
1703 self.hv_list = self.op.enabled_hypervisors
1704 if not self.hv_list:
1705 raise errors.OpPrereqError("Enabled hypervisors list must contain at"
1706 " least one member")
1707 invalid_hvs = set(self.hv_list) - constants.HYPER_TYPES
1709 raise errors.OpPrereqError("Enabled hypervisors contains invalid"
1710 " entries: %s" % invalid_hvs)
1712 self.hv_list = cluster.enabled_hypervisors
1714 if self.op.hvparams or self.op.enabled_hypervisors is not None:
1715 # either the enabled list has changed, or the parameters have, validate
1716 for hv_name, hv_params in self.new_hvparams.items():
1717 if ((self.op.hvparams and hv_name in self.op.hvparams) or
1718 (self.op.enabled_hypervisors and
1719 hv_name in self.op.enabled_hypervisors)):
1720 # either this is a new hypervisor, or its parameters have changed
1721 hv_class = hypervisor.GetHypervisor(hv_name)
1722 utils.ForceDictType(hv_params, constants.HVS_PARAMETER_TYPES)
1723 hv_class.CheckParameterSyntax(hv_params)
1724 _CheckHVParams(self, node_list, hv_name, hv_params)
1726 def Exec(self, feedback_fn):
1727 """Change the parameters of the cluster.
1730 if self.op.vg_name is not None:
1731 new_volume = self.op.vg_name
1734 if new_volume != self.cfg.GetVGName():
1735 self.cfg.SetVGName(new_volume)
1737 feedback_fn("Cluster LVM configuration already in desired"
1738 " state, not changing")
1739 if self.op.hvparams:
1740 self.cluster.hvparams = self.new_hvparams
1741 if self.op.enabled_hypervisors is not None:
1742 self.cluster.enabled_hypervisors = self.op.enabled_hypervisors
1743 if self.op.beparams:
1744 self.cluster.beparams[constants.PP_DEFAULT] = self.new_beparams
1745 if self.op.nicparams:
1746 self.cluster.nicparams[constants.PP_DEFAULT] = self.new_nicparams
1748 if self.op.candidate_pool_size is not None:
1749 self.cluster.candidate_pool_size = self.op.candidate_pool_size
1750 # we need to update the pool size here, otherwise the save will fail
1751 _AdjustCandidatePool(self)
1753 self.cfg.Update(self.cluster)
1756 def _RedistributeAncillaryFiles(lu, additional_nodes=None):
1757 """Distribute additional files which are part of the cluster configuration.
1759 ConfigWriter takes care of distributing the config and ssconf files, but
1760 there are more files which should be distributed to all nodes. This function
1761 makes sure those are copied.
1763 @param lu: calling logical unit
1764 @param additional_nodes: list of nodes not in the config to distribute to
1767 # 1. Gather target nodes
1768 myself = lu.cfg.GetNodeInfo(lu.cfg.GetMasterNode())
1769 dist_nodes = lu.cfg.GetNodeList()
1770 if additional_nodes is not None:
1771 dist_nodes.extend(additional_nodes)
1772 if myself.name in dist_nodes:
1773 dist_nodes.remove(myself.name)
1774 # 2. Gather files to distribute
1775 dist_files = set([constants.ETC_HOSTS,
1776 constants.SSH_KNOWN_HOSTS_FILE,
1777 constants.RAPI_CERT_FILE,
1778 constants.RAPI_USERS_FILE,
1779 constants.HMAC_CLUSTER_KEY,
1782 enabled_hypervisors = lu.cfg.GetClusterInfo().enabled_hypervisors
1783 for hv_name in enabled_hypervisors:
1784 hv_class = hypervisor.GetHypervisor(hv_name)
1785 dist_files.update(hv_class.GetAncillaryFiles())
1787 # 3. Perform the files upload
1788 for fname in dist_files:
1789 if os.path.exists(fname):
1790 result = lu.rpc.call_upload_file(dist_nodes, fname)
1791 for to_node, to_result in result.items():
1792 msg = to_result.fail_msg
1794 msg = ("Copy of file %s to node %s failed: %s" %
1795 (fname, to_node, msg))
1796 lu.proc.LogWarning(msg)
1799 class LURedistributeConfig(NoHooksLU):
1800 """Force the redistribution of cluster configuration.
1802 This is a very simple LU.
1808 def ExpandNames(self):
1809 self.needed_locks = {
1810 locking.LEVEL_NODE: locking.ALL_SET,
1812 self.share_locks[locking.LEVEL_NODE] = 1
1814 def CheckPrereq(self):
1815 """Check prerequisites.
1819 def Exec(self, feedback_fn):
1820 """Redistribute the configuration.
1823 self.cfg.Update(self.cfg.GetClusterInfo())
1824 _RedistributeAncillaryFiles(self)
1827 def _WaitForSync(lu, instance, oneshot=False, unlock=False):
1828 """Sleep and poll for an instance's disk to sync.
1831 if not instance.disks:
1835 lu.proc.LogInfo("Waiting for instance %s to sync disks." % instance.name)
1837 node = instance.primary_node
1839 for dev in instance.disks:
1840 lu.cfg.SetDiskID(dev, node)
1843 degr_retries = 10 # in seconds, as we sleep 1 second each time
1847 cumul_degraded = False
1848 rstats = lu.rpc.call_blockdev_getmirrorstatus(node, instance.disks)
1849 msg = rstats.fail_msg
1851 lu.LogWarning("Can't get any data from node %s: %s", node, msg)
1854 raise errors.RemoteError("Can't contact node %s for mirror data,"
1855 " aborting." % node)
1858 rstats = rstats.payload
1860 for i, mstat in enumerate(rstats):
1862 lu.LogWarning("Can't compute data for node %s/%s",
1863 node, instance.disks[i].iv_name)
1866 cumul_degraded = (cumul_degraded or
1867 (mstat.is_degraded and mstat.sync_percent is None))
1868 if mstat.sync_percent is not None:
1870 if mstat.estimated_time is not None:
1871 rem_time = "%d estimated seconds remaining" % mstat.estimated_time
1872 max_time = mstat.estimated_time
1874 rem_time = "no time estimate"
1875 lu.proc.LogInfo("- device %s: %5.2f%% done, %s" %
1876 (instance.disks[i].iv_name, mstat.sync_percent, rem_time))
1878 # if we're done but degraded, let's do a few small retries, to
1879 # make sure we see a stable and not transient situation; therefore
1880 # we force restart of the loop
1881 if (done or oneshot) and cumul_degraded and degr_retries > 0:
1882 logging.info("Degraded disks found, %d retries left", degr_retries)
1890 time.sleep(min(60, max_time))
1893 lu.proc.LogInfo("Instance %s's disks are in sync." % instance.name)
1894 return not cumul_degraded
1897 def _CheckDiskConsistency(lu, dev, node, on_primary, ldisk=False):
1898 """Check that mirrors are not degraded.
1900 The ldisk parameter, if True, will change the test from the
1901 is_degraded attribute (which represents overall non-ok status for
1902 the device(s)) to the ldisk (representing the local storage status).
1905 lu.cfg.SetDiskID(dev, node)
1909 if on_primary or dev.AssembleOnSecondary():
1910 rstats = lu.rpc.call_blockdev_find(node, dev)
1911 msg = rstats.fail_msg
1913 lu.LogWarning("Can't find disk on node %s: %s", node, msg)
1915 elif not rstats.payload:
1916 lu.LogWarning("Can't find disk on node %s", node)
1920 result = result and rstats.payload.ldisk_status == constants.LDS_OKAY
1922 result = result and not rstats.payload.is_degraded
1925 for child in dev.children:
1926 result = result and _CheckDiskConsistency(lu, child, node, on_primary)
1931 class LUDiagnoseOS(NoHooksLU):
1932 """Logical unit for OS diagnose/query.
1935 _OP_REQP = ["output_fields", "names"]
1937 _FIELDS_STATIC = utils.FieldSet()
1938 _FIELDS_DYNAMIC = utils.FieldSet("name", "valid", "node_status")
1940 def ExpandNames(self):
1942 raise errors.OpPrereqError("Selective OS query not supported")
1944 _CheckOutputFields(static=self._FIELDS_STATIC,
1945 dynamic=self._FIELDS_DYNAMIC,
1946 selected=self.op.output_fields)
1948 # Lock all nodes, in shared mode
1949 # Temporary removal of locks, should be reverted later
1950 # TODO: reintroduce locks when they are lighter-weight
1951 self.needed_locks = {}
1952 #self.share_locks[locking.LEVEL_NODE] = 1
1953 #self.needed_locks[locking.LEVEL_NODE] = locking.ALL_SET
1955 def CheckPrereq(self):
1956 """Check prerequisites.
1961 def _DiagnoseByOS(node_list, rlist):
1962 """Remaps a per-node return list into an a per-os per-node dictionary
1964 @param node_list: a list with the names of all nodes
1965 @param rlist: a map with node names as keys and OS objects as values
1968 @return: a dictionary with osnames as keys and as value another map, with
1969 nodes as keys and tuples of (path, status, diagnose) as values, eg::
1971 {"debian-etch": {"node1": [(/usr/lib/..., True, ""),
1972 (/srv/..., False, "invalid api")],
1973 "node2": [(/srv/..., True, "")]}
1978 # we build here the list of nodes that didn't fail the RPC (at RPC
1979 # level), so that nodes with a non-responding node daemon don't
1980 # make all OSes invalid
1981 good_nodes = [node_name for node_name in rlist
1982 if not rlist[node_name].fail_msg]
1983 for node_name, nr in rlist.items():
1984 if nr.fail_msg or not nr.payload:
1986 for name, path, status, diagnose in nr.payload:
1987 if name not in all_os:
1988 # build a list of nodes for this os containing empty lists
1989 # for each node in node_list
1991 for nname in good_nodes:
1992 all_os[name][nname] = []
1993 all_os[name][node_name].append((path, status, diagnose))
1996 def Exec(self, feedback_fn):
1997 """Compute the list of OSes.
2000 valid_nodes = [node for node in self.cfg.GetOnlineNodeList()]
2001 node_data = self.rpc.call_os_diagnose(valid_nodes)
2002 pol = self._DiagnoseByOS(valid_nodes, node_data)
2004 for os_name, os_data in pol.items():
2006 for field in self.op.output_fields:
2009 elif field == "valid":
2010 val = utils.all([osl and osl[0][1] for osl in os_data.values()])
2011 elif field == "node_status":
2012 # this is just a copy of the dict
2014 for node_name, nos_list in os_data.items():
2015 val[node_name] = nos_list
2017 raise errors.ParameterError(field)
2024 class LURemoveNode(LogicalUnit):
2025 """Logical unit for removing a node.
2028 HPATH = "node-remove"
2029 HTYPE = constants.HTYPE_NODE
2030 _OP_REQP = ["node_name"]
2032 def BuildHooksEnv(self):
2035 This doesn't run on the target node in the pre phase as a failed
2036 node would then be impossible to remove.
2040 "OP_TARGET": self.op.node_name,
2041 "NODE_NAME": self.op.node_name,
2043 all_nodes = self.cfg.GetNodeList()
2044 all_nodes.remove(self.op.node_name)
2045 return env, all_nodes, all_nodes
2047 def CheckPrereq(self):
2048 """Check prerequisites.
2051 - the node exists in the configuration
2052 - it does not have primary or secondary instances
2053 - it's not the master
2055 Any errors are signaled by raising errors.OpPrereqError.
2058 node = self.cfg.GetNodeInfo(self.cfg.ExpandNodeName(self.op.node_name))
2060 raise errors.OpPrereqError, ("Node '%s' is unknown." % self.op.node_name)
2062 instance_list = self.cfg.GetInstanceList()
2064 masternode = self.cfg.GetMasterNode()
2065 if node.name == masternode:
2066 raise errors.OpPrereqError("Node is the master node,"
2067 " you need to failover first.")
2069 for instance_name in instance_list:
2070 instance = self.cfg.GetInstanceInfo(instance_name)
2071 if node.name in instance.all_nodes:
2072 raise errors.OpPrereqError("Instance %s is still running on the node,"
2073 " please remove first." % instance_name)
2074 self.op.node_name = node.name
2077 def Exec(self, feedback_fn):
2078 """Removes the node from the cluster.
2082 logging.info("Stopping the node daemon and removing configs from node %s",
2085 self.context.RemoveNode(node.name)
2087 result = self.rpc.call_node_leave_cluster(node.name)
2088 msg = result.fail_msg
2090 self.LogWarning("Errors encountered on the remote node while leaving"
2091 " the cluster: %s", msg)
2093 # Promote nodes to master candidate as needed
2094 _AdjustCandidatePool(self)
2097 class LUQueryNodes(NoHooksLU):
2098 """Logical unit for querying nodes.
2101 _OP_REQP = ["output_fields", "names", "use_locking"]
2103 _FIELDS_DYNAMIC = utils.FieldSet(
2105 "mtotal", "mnode", "mfree",
2107 "ctotal", "cnodes", "csockets",
2110 _FIELDS_STATIC = utils.FieldSet(
2111 "name", "pinst_cnt", "sinst_cnt",
2112 "pinst_list", "sinst_list",
2113 "pip", "sip", "tags",
2122 def ExpandNames(self):
2123 _CheckOutputFields(static=self._FIELDS_STATIC,
2124 dynamic=self._FIELDS_DYNAMIC,
2125 selected=self.op.output_fields)
2127 self.needed_locks = {}
2128 self.share_locks[locking.LEVEL_NODE] = 1
2131 self.wanted = _GetWantedNodes(self, self.op.names)
2133 self.wanted = locking.ALL_SET
2135 self.do_node_query = self._FIELDS_STATIC.NonMatching(self.op.output_fields)
2136 self.do_locking = self.do_node_query and self.op.use_locking
2138 # if we don't request only static fields, we need to lock the nodes
2139 self.needed_locks[locking.LEVEL_NODE] = self.wanted
2142 def CheckPrereq(self):
2143 """Check prerequisites.
2146 # The validation of the node list is done in the _GetWantedNodes,
2147 # if non empty, and if empty, there's no validation to do
2150 def Exec(self, feedback_fn):
2151 """Computes the list of nodes and their attributes.
2154 all_info = self.cfg.GetAllNodesInfo()
2156 nodenames = self.acquired_locks[locking.LEVEL_NODE]
2157 elif self.wanted != locking.ALL_SET:
2158 nodenames = self.wanted
2159 missing = set(nodenames).difference(all_info.keys())
2161 raise errors.OpExecError(
2162 "Some nodes were removed before retrieving their data: %s" % missing)
2164 nodenames = all_info.keys()
2166 nodenames = utils.NiceSort(nodenames)
2167 nodelist = [all_info[name] for name in nodenames]
2169 # begin data gathering
2171 if self.do_node_query:
2173 node_data = self.rpc.call_node_info(nodenames, self.cfg.GetVGName(),
2174 self.cfg.GetHypervisorType())
2175 for name in nodenames:
2176 nodeinfo = node_data[name]
2177 if not nodeinfo.fail_msg and nodeinfo.payload:
2178 nodeinfo = nodeinfo.payload
2179 fn = utils.TryConvert
2181 "mtotal": fn(int, nodeinfo.get('memory_total', None)),
2182 "mnode": fn(int, nodeinfo.get('memory_dom0', None)),
2183 "mfree": fn(int, nodeinfo.get('memory_free', None)),
2184 "dtotal": fn(int, nodeinfo.get('vg_size', None)),
2185 "dfree": fn(int, nodeinfo.get('vg_free', None)),
2186 "ctotal": fn(int, nodeinfo.get('cpu_total', None)),
2187 "bootid": nodeinfo.get('bootid', None),
2188 "cnodes": fn(int, nodeinfo.get('cpu_nodes', None)),
2189 "csockets": fn(int, nodeinfo.get('cpu_sockets', None)),
2192 live_data[name] = {}
2194 live_data = dict.fromkeys(nodenames, {})
2196 node_to_primary = dict([(name, set()) for name in nodenames])
2197 node_to_secondary = dict([(name, set()) for name in nodenames])
2199 inst_fields = frozenset(("pinst_cnt", "pinst_list",
2200 "sinst_cnt", "sinst_list"))
2201 if inst_fields & frozenset(self.op.output_fields):
2202 instancelist = self.cfg.GetInstanceList()
2204 for instance_name in instancelist:
2205 inst = self.cfg.GetInstanceInfo(instance_name)
2206 if inst.primary_node in node_to_primary:
2207 node_to_primary[inst.primary_node].add(inst.name)
2208 for secnode in inst.secondary_nodes:
2209 if secnode in node_to_secondary:
2210 node_to_secondary[secnode].add(inst.name)
2212 master_node = self.cfg.GetMasterNode()
2214 # end data gathering
2217 for node in nodelist:
2219 for field in self.op.output_fields:
2222 elif field == "pinst_list":
2223 val = list(node_to_primary[node.name])
2224 elif field == "sinst_list":
2225 val = list(node_to_secondary[node.name])
2226 elif field == "pinst_cnt":
2227 val = len(node_to_primary[node.name])
2228 elif field == "sinst_cnt":
2229 val = len(node_to_secondary[node.name])
2230 elif field == "pip":
2231 val = node.primary_ip
2232 elif field == "sip":
2233 val = node.secondary_ip
2234 elif field == "tags":
2235 val = list(node.GetTags())
2236 elif field == "serial_no":
2237 val = node.serial_no
2238 elif field == "master_candidate":
2239 val = node.master_candidate
2240 elif field == "master":
2241 val = node.name == master_node
2242 elif field == "offline":
2244 elif field == "drained":
2246 elif self._FIELDS_DYNAMIC.Matches(field):
2247 val = live_data[node.name].get(field, None)
2248 elif field == "role":
2249 if node.name == master_node:
2251 elif node.master_candidate:
2260 raise errors.ParameterError(field)
2261 node_output.append(val)
2262 output.append(node_output)
2267 class LUQueryNodeVolumes(NoHooksLU):
2268 """Logical unit for getting volumes on node(s).
2271 _OP_REQP = ["nodes", "output_fields"]
2273 _FIELDS_DYNAMIC = utils.FieldSet("phys", "vg", "name", "size", "instance")
2274 _FIELDS_STATIC = utils.FieldSet("node")
2276 def ExpandNames(self):
2277 _CheckOutputFields(static=self._FIELDS_STATIC,
2278 dynamic=self._FIELDS_DYNAMIC,
2279 selected=self.op.output_fields)
2281 self.needed_locks = {}
2282 self.share_locks[locking.LEVEL_NODE] = 1
2283 if not self.op.nodes:
2284 self.needed_locks[locking.LEVEL_NODE] = locking.ALL_SET
2286 self.needed_locks[locking.LEVEL_NODE] = \
2287 _GetWantedNodes(self, self.op.nodes)
2289 def CheckPrereq(self):
2290 """Check prerequisites.
2292 This checks that the fields required are valid output fields.
2295 self.nodes = self.acquired_locks[locking.LEVEL_NODE]
2297 def Exec(self, feedback_fn):
2298 """Computes the list of nodes and their attributes.
2301 nodenames = self.nodes
2302 volumes = self.rpc.call_node_volumes(nodenames)
2304 ilist = [self.cfg.GetInstanceInfo(iname) for iname
2305 in self.cfg.GetInstanceList()]
2307 lv_by_node = dict([(inst, inst.MapLVsByNode()) for inst in ilist])
2310 for node in nodenames:
2311 nresult = volumes[node]
2314 msg = nresult.fail_msg
2316 self.LogWarning("Can't compute volume data on node %s: %s", node, msg)
2319 node_vols = nresult.payload[:]
2320 node_vols.sort(key=lambda vol: vol['dev'])
2322 for vol in node_vols:
2324 for field in self.op.output_fields:
2327 elif field == "phys":
2331 elif field == "name":
2333 elif field == "size":
2334 val = int(float(vol['size']))
2335 elif field == "instance":
2337 if node not in lv_by_node[inst]:
2339 if vol['name'] in lv_by_node[inst][node]:
2345 raise errors.ParameterError(field)
2346 node_output.append(str(val))
2348 output.append(node_output)
2353 class LUQueryNodeStorage(NoHooksLU):
2354 """Logical unit for getting information on storage units on node(s).
2357 _OP_REQP = ["nodes", "storage_type", "output_fields"]
2359 _FIELDS_STATIC = utils.FieldSet("node")
2361 def ExpandNames(self):
2362 storage_type = self.op.storage_type
2364 if storage_type not in constants.VALID_STORAGE_FIELDS:
2365 raise errors.OpPrereqError("Unknown storage type: %s" % storage_type)
2367 dynamic_fields = constants.VALID_STORAGE_FIELDS[storage_type]
2369 _CheckOutputFields(static=self._FIELDS_STATIC,
2370 dynamic=utils.FieldSet(*dynamic_fields),
2371 selected=self.op.output_fields)
2373 self.needed_locks = {}
2374 self.share_locks[locking.LEVEL_NODE] = 1
2377 self.needed_locks[locking.LEVEL_NODE] = \
2378 _GetWantedNodes(self, self.op.nodes)
2380 self.needed_locks[locking.LEVEL_NODE] = locking.ALL_SET
2382 def CheckPrereq(self):
2383 """Check prerequisites.
2385 This checks that the fields required are valid output fields.
2388 self.op.name = getattr(self.op, "name", None)
2390 self.nodes = self.acquired_locks[locking.LEVEL_NODE]
2392 def Exec(self, feedback_fn):
2393 """Computes the list of nodes and their attributes.
2396 # Always get name to sort by
2397 if constants.SF_NAME in self.op.output_fields:
2398 fields = self.op.output_fields[:]
2400 fields = [constants.SF_NAME] + self.op.output_fields
2402 # Never ask for node as it's only known to the LU
2403 while "node" in fields:
2404 fields.remove("node")
2406 field_idx = dict([(name, idx) for (idx, name) in enumerate(fields)])
2407 name_idx = field_idx[constants.SF_NAME]
2409 st_args = _GetStorageTypeArgs(self.cfg, self.op.storage_type)
2410 data = self.rpc.call_storage_list(self.nodes,
2411 self.op.storage_type, st_args,
2412 self.op.name, fields)
2416 for node in utils.NiceSort(self.nodes):
2417 nresult = data[node]
2421 msg = nresult.fail_msg
2423 self.LogWarning("Can't get storage data from node %s: %s", node, msg)
2426 rows = dict([(row[name_idx], row) for row in nresult.payload])
2428 for name in utils.NiceSort(rows.keys()):
2433 for field in self.op.output_fields:
2436 elif field in field_idx:
2437 val = row[field_idx[field]]
2439 raise errors.ParameterError(field)
2448 class LUModifyNodeStorage(NoHooksLU):
2449 """Logical unit for modifying a storage volume on a node.
2452 _OP_REQP = ["node_name", "storage_type", "name", "changes"]
2455 def CheckArguments(self):
2456 node_name = self.cfg.ExpandNodeName(self.op.node_name)
2457 if node_name is None:
2458 raise errors.OpPrereqError("Invalid node name '%s'" % self.op.node_name)
2460 self.op.node_name = node_name
2462 storage_type = self.op.storage_type
2463 if storage_type not in constants.VALID_STORAGE_FIELDS:
2464 raise errors.OpPrereqError("Unknown storage type: %s" % storage_type)
2466 def ExpandNames(self):
2467 self.needed_locks = {
2468 locking.LEVEL_NODE: self.op.node_name,
2471 def CheckPrereq(self):
2472 """Check prerequisites.
2475 storage_type = self.op.storage_type
2478 modifiable = constants.MODIFIABLE_STORAGE_FIELDS[storage_type]
2480 raise errors.OpPrereqError("Storage units of type '%s' can not be"
2481 " modified" % storage_type)
2483 diff = set(self.op.changes.keys()) - modifiable
2485 raise errors.OpPrereqError("The following fields can not be modified for"
2486 " storage units of type '%s': %r" %
2487 (storage_type, list(diff)))
2489 def Exec(self, feedback_fn):
2490 """Computes the list of nodes and their attributes.
2493 st_args = _GetStorageTypeArgs(self.cfg, self.op.storage_type)
2494 result = self.rpc.call_storage_modify(self.op.node_name,
2495 self.op.storage_type, st_args,
2496 self.op.name, self.op.changes)
2497 result.Raise("Failed to modify storage unit '%s' on %s" %
2498 (self.op.name, self.op.node_name))
2501 class LUAddNode(LogicalUnit):
2502 """Logical unit for adding node to the cluster.
2506 HTYPE = constants.HTYPE_NODE
2507 _OP_REQP = ["node_name"]
2509 def BuildHooksEnv(self):
2512 This will run on all nodes before, and on all nodes + the new node after.
2516 "OP_TARGET": self.op.node_name,
2517 "NODE_NAME": self.op.node_name,
2518 "NODE_PIP": self.op.primary_ip,
2519 "NODE_SIP": self.op.secondary_ip,
2521 nodes_0 = self.cfg.GetNodeList()
2522 nodes_1 = nodes_0 + [self.op.node_name, ]
2523 return env, nodes_0, nodes_1
2525 def CheckPrereq(self):
2526 """Check prerequisites.
2529 - the new node is not already in the config
2531 - its parameters (single/dual homed) matches the cluster
2533 Any errors are signaled by raising errors.OpPrereqError.
2536 node_name = self.op.node_name
2539 dns_data = utils.HostInfo(node_name)
2541 node = dns_data.name
2542 primary_ip = self.op.primary_ip = dns_data.ip
2543 secondary_ip = getattr(self.op, "secondary_ip", None)
2544 if secondary_ip is None:
2545 secondary_ip = primary_ip
2546 if not utils.IsValidIP(secondary_ip):
2547 raise errors.OpPrereqError("Invalid secondary IP given")
2548 self.op.secondary_ip = secondary_ip
2550 node_list = cfg.GetNodeList()
2551 if not self.op.readd and node in node_list:
2552 raise errors.OpPrereqError("Node %s is already in the configuration" %
2554 elif self.op.readd and node not in node_list:
2555 raise errors.OpPrereqError("Node %s is not in the configuration" % node)
2557 for existing_node_name in node_list:
2558 existing_node = cfg.GetNodeInfo(existing_node_name)
2560 if self.op.readd and node == existing_node_name:
2561 if (existing_node.primary_ip != primary_ip or
2562 existing_node.secondary_ip != secondary_ip):
2563 raise errors.OpPrereqError("Readded node doesn't have the same IP"
2564 " address configuration as before")
2567 if (existing_node.primary_ip == primary_ip or
2568 existing_node.secondary_ip == primary_ip or
2569 existing_node.primary_ip == secondary_ip or
2570 existing_node.secondary_ip == secondary_ip):
2571 raise errors.OpPrereqError("New node ip address(es) conflict with"
2572 " existing node %s" % existing_node.name)
2574 # check that the type of the node (single versus dual homed) is the
2575 # same as for the master
2576 myself = cfg.GetNodeInfo(self.cfg.GetMasterNode())
2577 master_singlehomed = myself.secondary_ip == myself.primary_ip
2578 newbie_singlehomed = secondary_ip == primary_ip
2579 if master_singlehomed != newbie_singlehomed:
2580 if master_singlehomed:
2581 raise errors.OpPrereqError("The master has no private ip but the"
2582 " new node has one")
2584 raise errors.OpPrereqError("The master has a private ip but the"
2585 " new node doesn't have one")
2587 # checks reachability
2588 if not utils.TcpPing(primary_ip, constants.DEFAULT_NODED_PORT):
2589 raise errors.OpPrereqError("Node not reachable by ping")
2591 if not newbie_singlehomed:
2592 # check reachability from my secondary ip to newbie's secondary ip
2593 if not utils.TcpPing(secondary_ip, constants.DEFAULT_NODED_PORT,
2594 source=myself.secondary_ip):
2595 raise errors.OpPrereqError("Node secondary ip not reachable by TCP"
2596 " based ping to noded port")
2598 cp_size = self.cfg.GetClusterInfo().candidate_pool_size
2603 mc_now, mc_max = self.cfg.GetMasterCandidateStats(exceptions)
2604 # the new node will increase mc_max with one, so:
2605 mc_max = min(mc_max + 1, cp_size)
2606 self.master_candidate = mc_now < mc_max
2609 self.new_node = self.cfg.GetNodeInfo(node)
2610 assert self.new_node is not None, "Can't retrieve locked node %s" % node
2612 self.new_node = objects.Node(name=node,
2613 primary_ip=primary_ip,
2614 secondary_ip=secondary_ip,
2615 master_candidate=self.master_candidate,
2616 offline=False, drained=False)
2618 def Exec(self, feedback_fn):
2619 """Adds the new node to the cluster.
2622 new_node = self.new_node
2623 node = new_node.name
2625 # for re-adds, reset the offline/drained/master-candidate flags;
2626 # we need to reset here, otherwise offline would prevent RPC calls
2627 # later in the procedure; this also means that if the re-add
2628 # fails, we are left with a non-offlined, broken node
2630 new_node.drained = new_node.offline = False
2631 self.LogInfo("Readding a node, the offline/drained flags were reset")
2632 # if we demote the node, we do cleanup later in the procedure
2633 new_node.master_candidate = self.master_candidate
2635 # notify the user about any possible mc promotion
2636 if new_node.master_candidate:
2637 self.LogInfo("Node will be a master candidate")
2639 # check connectivity
2640 result = self.rpc.call_version([node])[node]
2641 result.Raise("Can't get version information from node %s" % node)
2642 if constants.PROTOCOL_VERSION == result.payload:
2643 logging.info("Communication to node %s fine, sw version %s match",
2644 node, result.payload)
2646 raise errors.OpExecError("Version mismatch master version %s,"
2647 " node version %s" %
2648 (constants.PROTOCOL_VERSION, result.payload))
2651 logging.info("Copy ssh key to node %s", node)
2652 priv_key, pub_key, _ = ssh.GetUserFiles(constants.GANETI_RUNAS)
2654 keyfiles = [constants.SSH_HOST_DSA_PRIV, constants.SSH_HOST_DSA_PUB,
2655 constants.SSH_HOST_RSA_PRIV, constants.SSH_HOST_RSA_PUB,
2661 keyarray.append(f.read())
2665 result = self.rpc.call_node_add(node, keyarray[0], keyarray[1],
2667 keyarray[3], keyarray[4], keyarray[5])
2668 result.Raise("Cannot transfer ssh keys to the new node")
2670 # Add node to our /etc/hosts, and add key to known_hosts
2671 if self.cfg.GetClusterInfo().modify_etc_hosts:
2672 utils.AddHostToEtcHosts(new_node.name)
2674 if new_node.secondary_ip != new_node.primary_ip:
2675 result = self.rpc.call_node_has_ip_address(new_node.name,
2676 new_node.secondary_ip)
2677 result.Raise("Failure checking secondary ip on node %s" % new_node.name,
2679 if not result.payload:
2680 raise errors.OpExecError("Node claims it doesn't have the secondary ip"
2681 " you gave (%s). Please fix and re-run this"
2682 " command." % new_node.secondary_ip)
2684 node_verify_list = [self.cfg.GetMasterNode()]
2685 node_verify_param = {
2687 # TODO: do a node-net-test as well?
2690 result = self.rpc.call_node_verify(node_verify_list, node_verify_param,
2691 self.cfg.GetClusterName())
2692 for verifier in node_verify_list:
2693 result[verifier].Raise("Cannot communicate with node %s" % verifier)
2694 nl_payload = result[verifier].payload['nodelist']
2696 for failed in nl_payload:
2697 feedback_fn("ssh/hostname verification failed %s -> %s" %
2698 (verifier, nl_payload[failed]))
2699 raise errors.OpExecError("ssh/hostname verification failed.")
2702 _RedistributeAncillaryFiles(self)
2703 self.context.ReaddNode(new_node)
2704 # make sure we redistribute the config
2705 self.cfg.Update(new_node)
2706 # and make sure the new node will not have old files around
2707 if not new_node.master_candidate:
2708 result = self.rpc.call_node_demote_from_mc(new_node.name)
2709 msg = result.RemoteFailMsg()
2711 self.LogWarning("Node failed to demote itself from master"
2712 " candidate status: %s" % msg)
2714 _RedistributeAncillaryFiles(self, additional_nodes=[node])
2715 self.context.AddNode(new_node)
2718 class LUSetNodeParams(LogicalUnit):
2719 """Modifies the parameters of a node.
2722 HPATH = "node-modify"
2723 HTYPE = constants.HTYPE_NODE
2724 _OP_REQP = ["node_name"]
2727 def CheckArguments(self):
2728 node_name = self.cfg.ExpandNodeName(self.op.node_name)
2729 if node_name is None:
2730 raise errors.OpPrereqError("Invalid node name '%s'" % self.op.node_name)
2731 self.op.node_name = node_name
2732 _CheckBooleanOpField(self.op, 'master_candidate')
2733 _CheckBooleanOpField(self.op, 'offline')
2734 _CheckBooleanOpField(self.op, 'drained')
2735 all_mods = [self.op.offline, self.op.master_candidate, self.op.drained]
2736 if all_mods.count(None) == 3:
2737 raise errors.OpPrereqError("Please pass at least one modification")
2738 if all_mods.count(True) > 1:
2739 raise errors.OpPrereqError("Can't set the node into more than one"
2740 " state at the same time")
2742 def ExpandNames(self):
2743 self.needed_locks = {locking.LEVEL_NODE: self.op.node_name}
2745 def BuildHooksEnv(self):
2748 This runs on the master node.
2752 "OP_TARGET": self.op.node_name,
2753 "MASTER_CANDIDATE": str(self.op.master_candidate),
2754 "OFFLINE": str(self.op.offline),
2755 "DRAINED": str(self.op.drained),
2757 nl = [self.cfg.GetMasterNode(),
2761 def CheckPrereq(self):
2762 """Check prerequisites.
2764 This only checks the instance list against the existing names.
2767 node = self.node = self.cfg.GetNodeInfo(self.op.node_name)
2769 if ((self.op.master_candidate == False or self.op.offline == True or
2770 self.op.drained == True) and node.master_candidate):
2771 # we will demote the node from master_candidate
2772 if self.op.node_name == self.cfg.GetMasterNode():
2773 raise errors.OpPrereqError("The master node has to be a"
2774 " master candidate, online and not drained")
2775 cp_size = self.cfg.GetClusterInfo().candidate_pool_size
2776 num_candidates, _ = self.cfg.GetMasterCandidateStats()
2777 if num_candidates <= cp_size:
2778 msg = ("Not enough master candidates (desired"
2779 " %d, new value will be %d)" % (cp_size, num_candidates-1))
2781 self.LogWarning(msg)
2783 raise errors.OpPrereqError(msg)
2785 if (self.op.master_candidate == True and
2786 ((node.offline and not self.op.offline == False) or
2787 (node.drained and not self.op.drained == False))):
2788 raise errors.OpPrereqError("Node '%s' is offline or drained, can't set"
2789 " to master_candidate" % node.name)
2793 def Exec(self, feedback_fn):
2802 if self.op.offline is not None:
2803 node.offline = self.op.offline
2804 result.append(("offline", str(self.op.offline)))
2805 if self.op.offline == True:
2806 if node.master_candidate:
2807 node.master_candidate = False
2809 result.append(("master_candidate", "auto-demotion due to offline"))
2811 node.drained = False
2812 result.append(("drained", "clear drained status due to offline"))
2814 if self.op.master_candidate is not None:
2815 node.master_candidate = self.op.master_candidate
2817 result.append(("master_candidate", str(self.op.master_candidate)))
2818 if self.op.master_candidate == False:
2819 rrc = self.rpc.call_node_demote_from_mc(node.name)
2822 self.LogWarning("Node failed to demote itself: %s" % msg)
2824 if self.op.drained is not None:
2825 node.drained = self.op.drained
2826 result.append(("drained", str(self.op.drained)))
2827 if self.op.drained == True:
2828 if node.master_candidate:
2829 node.master_candidate = False
2831 result.append(("master_candidate", "auto-demotion due to drain"))
2832 rrc = self.rpc.call_node_demote_from_mc(node.name)
2833 msg = rrc.RemoteFailMsg()
2835 self.LogWarning("Node failed to demote itself: %s" % msg)
2837 node.offline = False
2838 result.append(("offline", "clear offline status due to drain"))
2840 # this will trigger configuration file update, if needed
2841 self.cfg.Update(node)
2842 # this will trigger job queue propagation or cleanup
2844 self.context.ReaddNode(node)
2849 class LUPowercycleNode(NoHooksLU):
2850 """Powercycles a node.
2853 _OP_REQP = ["node_name", "force"]
2856 def CheckArguments(self):
2857 node_name = self.cfg.ExpandNodeName(self.op.node_name)
2858 if node_name is None:
2859 raise errors.OpPrereqError("Invalid node name '%s'" % self.op.node_name)
2860 self.op.node_name = node_name
2861 if node_name == self.cfg.GetMasterNode() and not self.op.force:
2862 raise errors.OpPrereqError("The node is the master and the force"
2863 " parameter was not set")
2865 def ExpandNames(self):
2866 """Locking for PowercycleNode.
2868 This is a last-resort option and shouldn't block on other
2869 jobs. Therefore, we grab no locks.
2872 self.needed_locks = {}
2874 def CheckPrereq(self):
2875 """Check prerequisites.
2877 This LU has no prereqs.
2882 def Exec(self, feedback_fn):
2886 result = self.rpc.call_node_powercycle(self.op.node_name,
2887 self.cfg.GetHypervisorType())
2888 result.Raise("Failed to schedule the reboot")
2889 return result.payload
2892 class LUQueryClusterInfo(NoHooksLU):
2893 """Query cluster configuration.
2899 def ExpandNames(self):
2900 self.needed_locks = {}
2902 def CheckPrereq(self):
2903 """No prerequsites needed for this LU.
2908 def Exec(self, feedback_fn):
2909 """Return cluster config.
2912 cluster = self.cfg.GetClusterInfo()
2914 "software_version": constants.RELEASE_VERSION,
2915 "protocol_version": constants.PROTOCOL_VERSION,
2916 "config_version": constants.CONFIG_VERSION,
2917 "os_api_version": max(constants.OS_API_VERSIONS),
2918 "export_version": constants.EXPORT_VERSION,
2919 "architecture": (platform.architecture()[0], platform.machine()),
2920 "name": cluster.cluster_name,
2921 "master": cluster.master_node,
2922 "default_hypervisor": cluster.enabled_hypervisors[0],
2923 "enabled_hypervisors": cluster.enabled_hypervisors,
2924 "hvparams": dict([(hypervisor_name, cluster.hvparams[hypervisor_name])
2925 for hypervisor_name in cluster.enabled_hypervisors]),
2926 "beparams": cluster.beparams,
2927 "nicparams": cluster.nicparams,
2928 "candidate_pool_size": cluster.candidate_pool_size,
2929 "master_netdev": cluster.master_netdev,
2930 "volume_group_name": cluster.volume_group_name,
2931 "file_storage_dir": cluster.file_storage_dir,
2937 class LUQueryConfigValues(NoHooksLU):
2938 """Return configuration values.
2943 _FIELDS_DYNAMIC = utils.FieldSet()
2944 _FIELDS_STATIC = utils.FieldSet("cluster_name", "master_node", "drain_flag")
2946 def ExpandNames(self):
2947 self.needed_locks = {}
2949 _CheckOutputFields(static=self._FIELDS_STATIC,
2950 dynamic=self._FIELDS_DYNAMIC,
2951 selected=self.op.output_fields)
2953 def CheckPrereq(self):
2954 """No prerequisites.
2959 def Exec(self, feedback_fn):
2960 """Dump a representation of the cluster config to the standard output.
2964 for field in self.op.output_fields:
2965 if field == "cluster_name":
2966 entry = self.cfg.GetClusterName()
2967 elif field == "master_node":
2968 entry = self.cfg.GetMasterNode()
2969 elif field == "drain_flag":
2970 entry = os.path.exists(constants.JOB_QUEUE_DRAIN_FILE)
2972 raise errors.ParameterError(field)
2973 values.append(entry)
2977 class LUActivateInstanceDisks(NoHooksLU):
2978 """Bring up an instance's disks.
2981 _OP_REQP = ["instance_name"]
2984 def ExpandNames(self):
2985 self._ExpandAndLockInstance()
2986 self.needed_locks[locking.LEVEL_NODE] = []
2987 self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
2989 def DeclareLocks(self, level):
2990 if level == locking.LEVEL_NODE:
2991 self._LockInstancesNodes()
2993 def CheckPrereq(self):
2994 """Check prerequisites.
2996 This checks that the instance is in the cluster.
2999 self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
3000 assert self.instance is not None, \
3001 "Cannot retrieve locked instance %s" % self.op.instance_name
3002 _CheckNodeOnline(self, self.instance.primary_node)
3004 def Exec(self, feedback_fn):
3005 """Activate the disks.
3008 disks_ok, disks_info = _AssembleInstanceDisks(self, self.instance)
3010 raise errors.OpExecError("Cannot activate block devices")
3015 def _AssembleInstanceDisks(lu, instance, ignore_secondaries=False):
3016 """Prepare the block devices for an instance.
3018 This sets up the block devices on all nodes.
3020 @type lu: L{LogicalUnit}
3021 @param lu: the logical unit on whose behalf we execute
3022 @type instance: L{objects.Instance}
3023 @param instance: the instance for whose disks we assemble
3024 @type ignore_secondaries: boolean
3025 @param ignore_secondaries: if true, errors on secondary nodes
3026 won't result in an error return from the function
3027 @return: False if the operation failed, otherwise a list of
3028 (host, instance_visible_name, node_visible_name)
3029 with the mapping from node devices to instance devices
3034 iname = instance.name
3035 # With the two passes mechanism we try to reduce the window of
3036 # opportunity for the race condition of switching DRBD to primary
3037 # before handshaking occured, but we do not eliminate it
3039 # The proper fix would be to wait (with some limits) until the
3040 # connection has been made and drbd transitions from WFConnection
3041 # into any other network-connected state (Connected, SyncTarget,
3044 # 1st pass, assemble on all nodes in secondary mode
3045 for inst_disk in instance.disks:
3046 for node, node_disk in inst_disk.ComputeNodeTree(instance.primary_node):
3047 lu.cfg.SetDiskID(node_disk, node)
3048 result = lu.rpc.call_blockdev_assemble(node, node_disk, iname, False)
3049 msg = result.fail_msg
3051 lu.proc.LogWarning("Could not prepare block device %s on node %s"
3052 " (is_primary=False, pass=1): %s",
3053 inst_disk.iv_name, node, msg)
3054 if not ignore_secondaries:
3057 # FIXME: race condition on drbd migration to primary
3059 # 2nd pass, do only the primary node
3060 for inst_disk in instance.disks:
3061 for node, node_disk in inst_disk.ComputeNodeTree(instance.primary_node):
3062 if node != instance.primary_node:
3064 lu.cfg.SetDiskID(node_disk, node)
3065 result = lu.rpc.call_blockdev_assemble(node, node_disk, iname, True)
3066 msg = result.fail_msg
3068 lu.proc.LogWarning("Could not prepare block device %s on node %s"
3069 " (is_primary=True, pass=2): %s",
3070 inst_disk.iv_name, node, msg)
3072 device_info.append((instance.primary_node, inst_disk.iv_name,
3075 # leave the disks configured for the primary node
3076 # this is a workaround that would be fixed better by
3077 # improving the logical/physical id handling
3078 for disk in instance.disks:
3079 lu.cfg.SetDiskID(disk, instance.primary_node)
3081 return disks_ok, device_info
3084 def _StartInstanceDisks(lu, instance, force):
3085 """Start the disks of an instance.
3088 disks_ok, _ = _AssembleInstanceDisks(lu, instance,
3089 ignore_secondaries=force)
3091 _ShutdownInstanceDisks(lu, instance)
3092 if force is not None and not force:
3093 lu.proc.LogWarning("", hint="If the message above refers to a"
3095 " you can retry the operation using '--force'.")
3096 raise errors.OpExecError("Disk consistency error")
3099 class LUDeactivateInstanceDisks(NoHooksLU):
3100 """Shutdown an instance's disks.
3103 _OP_REQP = ["instance_name"]
3106 def ExpandNames(self):
3107 self._ExpandAndLockInstance()
3108 self.needed_locks[locking.LEVEL_NODE] = []
3109 self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
3111 def DeclareLocks(self, level):
3112 if level == locking.LEVEL_NODE:
3113 self._LockInstancesNodes()
3115 def CheckPrereq(self):
3116 """Check prerequisites.
3118 This checks that the instance is in the cluster.
3121 self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
3122 assert self.instance is not None, \
3123 "Cannot retrieve locked instance %s" % self.op.instance_name
3125 def Exec(self, feedback_fn):
3126 """Deactivate the disks
3129 instance = self.instance
3130 _SafeShutdownInstanceDisks(self, instance)
3133 def _SafeShutdownInstanceDisks(lu, instance):
3134 """Shutdown block devices of an instance.
3136 This function checks if an instance is running, before calling
3137 _ShutdownInstanceDisks.
3140 pnode = instance.primary_node
3141 ins_l = lu.rpc.call_instance_list([pnode], [instance.hypervisor])[pnode]
3142 ins_l.Raise("Can't contact node %s" % pnode)
3144 if instance.name in ins_l.payload:
3145 raise errors.OpExecError("Instance is running, can't shutdown"
3148 _ShutdownInstanceDisks(lu, instance)
3151 def _ShutdownInstanceDisks(lu, instance, ignore_primary=False):
3152 """Shutdown block devices of an instance.
3154 This does the shutdown on all nodes of the instance.
3156 If the ignore_primary is false, errors on the primary node are
3161 for disk in instance.disks:
3162 for node, top_disk in disk.ComputeNodeTree(instance.primary_node):
3163 lu.cfg.SetDiskID(top_disk, node)
3164 result = lu.rpc.call_blockdev_shutdown(node, top_disk)
3165 msg = result.fail_msg
3167 lu.LogWarning("Could not shutdown block device %s on node %s: %s",
3168 disk.iv_name, node, msg)
3169 if not ignore_primary or node != instance.primary_node:
3174 def _CheckNodeFreeMemory(lu, node, reason, requested, hypervisor_name):
3175 """Checks if a node has enough free memory.
3177 This function check if a given node has the needed amount of free
3178 memory. In case the node has less memory or we cannot get the
3179 information from the node, this function raise an OpPrereqError
3182 @type lu: C{LogicalUnit}
3183 @param lu: a logical unit from which we get configuration data
3185 @param node: the node to check
3186 @type reason: C{str}
3187 @param reason: string to use in the error message
3188 @type requested: C{int}
3189 @param requested: the amount of memory in MiB to check for
3190 @type hypervisor_name: C{str}
3191 @param hypervisor_name: the hypervisor to ask for memory stats
3192 @raise errors.OpPrereqError: if the node doesn't have enough memory, or
3193 we cannot check the node
3196 nodeinfo = lu.rpc.call_node_info([node], lu.cfg.GetVGName(), hypervisor_name)
3197 nodeinfo[node].Raise("Can't get data from node %s" % node, prereq=True)
3198 free_mem = nodeinfo[node].payload.get('memory_free', None)
3199 if not isinstance(free_mem, int):
3200 raise errors.OpPrereqError("Can't compute free memory on node %s, result"
3201 " was '%s'" % (node, free_mem))
3202 if requested > free_mem:
3203 raise errors.OpPrereqError("Not enough memory on node %s for %s:"
3204 " needed %s MiB, available %s MiB" %
3205 (node, reason, requested, free_mem))
3208 class LUStartupInstance(LogicalUnit):
3209 """Starts an instance.
3212 HPATH = "instance-start"
3213 HTYPE = constants.HTYPE_INSTANCE
3214 _OP_REQP = ["instance_name", "force"]
3217 def ExpandNames(self):
3218 self._ExpandAndLockInstance()
3220 def BuildHooksEnv(self):
3223 This runs on master, primary and secondary nodes of the instance.
3227 "FORCE": self.op.force,
3229 env.update(_BuildInstanceHookEnvByObject(self, self.instance))
3230 nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
3233 def CheckPrereq(self):
3234 """Check prerequisites.
3236 This checks that the instance is in the cluster.
3239 self.instance = instance = self.cfg.GetInstanceInfo(self.op.instance_name)
3240 assert self.instance is not None, \
3241 "Cannot retrieve locked instance %s" % self.op.instance_name
3244 self.beparams = getattr(self.op, "beparams", {})
3246 if not isinstance(self.beparams, dict):
3247 raise errors.OpPrereqError("Invalid beparams passed: %s, expected"
3248 " dict" % (type(self.beparams), ))
3249 # fill the beparams dict
3250 utils.ForceDictType(self.beparams, constants.BES_PARAMETER_TYPES)
3251 self.op.beparams = self.beparams
3254 self.hvparams = getattr(self.op, "hvparams", {})
3256 if not isinstance(self.hvparams, dict):
3257 raise errors.OpPrereqError("Invalid hvparams passed: %s, expected"
3258 " dict" % (type(self.hvparams), ))
3260 # check hypervisor parameter syntax (locally)
3261 cluster = self.cfg.GetClusterInfo()
3262 utils.ForceDictType(self.hvparams, constants.HVS_PARAMETER_TYPES)
3263 filled_hvp = objects.FillDict(cluster.hvparams[instance.hypervisor],
3265 filled_hvp.update(self.hvparams)
3266 hv_type = hypervisor.GetHypervisor(instance.hypervisor)
3267 hv_type.CheckParameterSyntax(filled_hvp)
3268 _CheckHVParams(self, instance.all_nodes, instance.hypervisor, filled_hvp)
3269 self.op.hvparams = self.hvparams
3271 _CheckNodeOnline(self, instance.primary_node)
3273 bep = self.cfg.GetClusterInfo().FillBE(instance)
3274 # check bridges existence
3275 _CheckInstanceBridgesExist(self, instance)
3277 remote_info = self.rpc.call_instance_info(instance.primary_node,
3279 instance.hypervisor)
3280 remote_info.Raise("Error checking node %s" % instance.primary_node,
3282 if not remote_info.payload: # not running already
3283 _CheckNodeFreeMemory(self, instance.primary_node,
3284 "starting instance %s" % instance.name,
3285 bep[constants.BE_MEMORY], instance.hypervisor)
3287 def Exec(self, feedback_fn):
3288 """Start the instance.
3291 instance = self.instance
3292 force = self.op.force
3294 self.cfg.MarkInstanceUp(instance.name)
3296 node_current = instance.primary_node
3298 _StartInstanceDisks(self, instance, force)
3300 result = self.rpc.call_instance_start(node_current, instance,
3301 self.hvparams, self.beparams)
3302 msg = result.fail_msg
3304 _ShutdownInstanceDisks(self, instance)
3305 raise errors.OpExecError("Could not start instance: %s" % msg)
3308 class LURebootInstance(LogicalUnit):
3309 """Reboot an instance.
3312 HPATH = "instance-reboot"
3313 HTYPE = constants.HTYPE_INSTANCE
3314 _OP_REQP = ["instance_name", "ignore_secondaries", "reboot_type"]
3317 def ExpandNames(self):
3318 if self.op.reboot_type not in [constants.INSTANCE_REBOOT_SOFT,
3319 constants.INSTANCE_REBOOT_HARD,
3320 constants.INSTANCE_REBOOT_FULL]:
3321 raise errors.ParameterError("reboot type not in [%s, %s, %s]" %
3322 (constants.INSTANCE_REBOOT_SOFT,
3323 constants.INSTANCE_REBOOT_HARD,
3324 constants.INSTANCE_REBOOT_FULL))
3325 self._ExpandAndLockInstance()
3327 def BuildHooksEnv(self):
3330 This runs on master, primary and secondary nodes of the instance.
3334 "IGNORE_SECONDARIES": self.op.ignore_secondaries,
3335 "REBOOT_TYPE": self.op.reboot_type,
3337 env.update(_BuildInstanceHookEnvByObject(self, self.instance))
3338 nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
3341 def CheckPrereq(self):
3342 """Check prerequisites.
3344 This checks that the instance is in the cluster.
3347 self.instance = instance = self.cfg.GetInstanceInfo(self.op.instance_name)
3348 assert self.instance is not None, \
3349 "Cannot retrieve locked instance %s" % self.op.instance_name
3351 _CheckNodeOnline(self, instance.primary_node)
3353 # check bridges existence
3354 _CheckInstanceBridgesExist(self, instance)
3356 def Exec(self, feedback_fn):
3357 """Reboot the instance.
3360 instance = self.instance
3361 ignore_secondaries = self.op.ignore_secondaries
3362 reboot_type = self.op.reboot_type
3364 node_current = instance.primary_node
3366 if reboot_type in [constants.INSTANCE_REBOOT_SOFT,
3367 constants.INSTANCE_REBOOT_HARD]:
3368 for disk in instance.disks:
3369 self.cfg.SetDiskID(disk, node_current)
3370 result = self.rpc.call_instance_reboot(node_current, instance,
3372 result.Raise("Could not reboot instance")
3374 result = self.rpc.call_instance_shutdown(node_current, instance)
3375 result.Raise("Could not shutdown instance for full reboot")
3376 _ShutdownInstanceDisks(self, instance)
3377 _StartInstanceDisks(self, instance, ignore_secondaries)
3378 result = self.rpc.call_instance_start(node_current, instance, None, None)
3379 msg = result.fail_msg
3381 _ShutdownInstanceDisks(self, instance)
3382 raise errors.OpExecError("Could not start instance for"
3383 " full reboot: %s" % msg)
3385 self.cfg.MarkInstanceUp(instance.name)
3388 class LUShutdownInstance(LogicalUnit):
3389 """Shutdown an instance.
3392 HPATH = "instance-stop"
3393 HTYPE = constants.HTYPE_INSTANCE
3394 _OP_REQP = ["instance_name"]
3397 def ExpandNames(self):
3398 self._ExpandAndLockInstance()
3400 def BuildHooksEnv(self):
3403 This runs on master, primary and secondary nodes of the instance.
3406 env = _BuildInstanceHookEnvByObject(self, self.instance)
3407 nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
3410 def CheckPrereq(self):
3411 """Check prerequisites.
3413 This checks that the instance is in the cluster.
3416 self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
3417 assert self.instance is not None, \
3418 "Cannot retrieve locked instance %s" % self.op.instance_name
3419 _CheckNodeOnline(self, self.instance.primary_node)
3421 def Exec(self, feedback_fn):
3422 """Shutdown the instance.
3425 instance = self.instance
3426 node_current = instance.primary_node
3427 self.cfg.MarkInstanceDown(instance.name)
3428 result = self.rpc.call_instance_shutdown(node_current, instance)
3429 msg = result.fail_msg
3431 self.proc.LogWarning("Could not shutdown instance: %s" % msg)
3433 _ShutdownInstanceDisks(self, instance)
3436 class LUReinstallInstance(LogicalUnit):
3437 """Reinstall an instance.
3440 HPATH = "instance-reinstall"
3441 HTYPE = constants.HTYPE_INSTANCE
3442 _OP_REQP = ["instance_name"]
3445 def ExpandNames(self):
3446 self._ExpandAndLockInstance()
3448 def BuildHooksEnv(self):
3451 This runs on master, primary and secondary nodes of the instance.
3454 env = _BuildInstanceHookEnvByObject(self, self.instance)
3455 nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
3458 def CheckPrereq(self):
3459 """Check prerequisites.
3461 This checks that the instance is in the cluster and is not running.
3464 instance = self.cfg.GetInstanceInfo(self.op.instance_name)
3465 assert instance is not None, \
3466 "Cannot retrieve locked instance %s" % self.op.instance_name
3467 _CheckNodeOnline(self, instance.primary_node)
3469 if instance.disk_template == constants.DT_DISKLESS:
3470 raise errors.OpPrereqError("Instance '%s' has no disks" %
3471 self.op.instance_name)
3472 if instance.admin_up:
3473 raise errors.OpPrereqError("Instance '%s' is marked to be up" %
3474 self.op.instance_name)
3475 remote_info = self.rpc.call_instance_info(instance.primary_node,
3477 instance.hypervisor)
3478 remote_info.Raise("Error checking node %s" % instance.primary_node,
3480 if remote_info.payload:
3481 raise errors.OpPrereqError("Instance '%s' is running on the node %s" %
3482 (self.op.instance_name,
3483 instance.primary_node))
3485 self.op.os_type = getattr(self.op, "os_type", None)
3486 if self.op.os_type is not None:
3488 pnode = self.cfg.GetNodeInfo(
3489 self.cfg.ExpandNodeName(instance.primary_node))
3491 raise errors.OpPrereqError("Primary node '%s' is unknown" %
3493 result = self.rpc.call_os_get(pnode.name, self.op.os_type)
3494 result.Raise("OS '%s' not in supported OS list for primary node %s" %
3495 (self.op.os_type, pnode.name), prereq=True)
3497 self.instance = instance
3499 def Exec(self, feedback_fn):
3500 """Reinstall the instance.
3503 inst = self.instance
3505 if self.op.os_type is not None:
3506 feedback_fn("Changing OS to '%s'..." % self.op.os_type)
3507 inst.os = self.op.os_type
3508 self.cfg.Update(inst)
3510 _StartInstanceDisks(self, inst, None)
3512 feedback_fn("Running the instance OS create scripts...")
3513 result = self.rpc.call_instance_os_add(inst.primary_node, inst, True)
3514 result.Raise("Could not install OS for instance %s on node %s" %
3515 (inst.name, inst.primary_node))
3517 _ShutdownInstanceDisks(self, inst)
3520 class LURenameInstance(LogicalUnit):
3521 """Rename an instance.
3524 HPATH = "instance-rename"
3525 HTYPE = constants.HTYPE_INSTANCE
3526 _OP_REQP = ["instance_name", "new_name"]
3528 def BuildHooksEnv(self):
3531 This runs on master, primary and secondary nodes of the instance.
3534 env = _BuildInstanceHookEnvByObject(self, self.instance)
3535 env["INSTANCE_NEW_NAME"] = self.op.new_name
3536 nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
3539 def CheckPrereq(self):
3540 """Check prerequisites.
3542 This checks that the instance is in the cluster and is not running.
3545 instance = self.cfg.GetInstanceInfo(
3546 self.cfg.ExpandInstanceName(self.op.instance_name))
3547 if instance is None:
3548 raise errors.OpPrereqError("Instance '%s' not known" %
3549 self.op.instance_name)
3550 _CheckNodeOnline(self, instance.primary_node)
3552 if instance.admin_up:
3553 raise errors.OpPrereqError("Instance '%s' is marked to be up" %
3554 self.op.instance_name)
3555 remote_info = self.rpc.call_instance_info(instance.primary_node,
3557 instance.hypervisor)
3558 remote_info.Raise("Error checking node %s" % instance.primary_node,
3560 if remote_info.payload:
3561 raise errors.OpPrereqError("Instance '%s' is running on the node %s" %
3562 (self.op.instance_name,
3563 instance.primary_node))
3564 self.instance = instance
3566 # new name verification
3567 name_info = utils.HostInfo(self.op.new_name)
3569 self.op.new_name = new_name = name_info.name
3570 instance_list = self.cfg.GetInstanceList()
3571 if new_name in instance_list:
3572 raise errors.OpPrereqError("Instance '%s' is already in the cluster" %
3575 if not getattr(self.op, "ignore_ip", False):
3576 if utils.TcpPing(name_info.ip, constants.DEFAULT_NODED_PORT):
3577 raise errors.OpPrereqError("IP %s of instance %s already in use" %
3578 (name_info.ip, new_name))
3581 def Exec(self, feedback_fn):
3582 """Reinstall the instance.
3585 inst = self.instance
3586 old_name = inst.name
3588 if inst.disk_template == constants.DT_FILE:
3589 old_file_storage_dir = os.path.dirname(inst.disks[0].logical_id[1])
3591 self.cfg.RenameInstance(inst.name, self.op.new_name)
3592 # Change the instance lock. This is definitely safe while we hold the BGL
3593 self.context.glm.remove(locking.LEVEL_INSTANCE, old_name)
3594 self.context.glm.add(locking.LEVEL_INSTANCE, self.op.new_name)
3596 # re-read the instance from the configuration after rename
3597 inst = self.cfg.GetInstanceInfo(self.op.new_name)
3599 if inst.disk_template == constants.DT_FILE:
3600 new_file_storage_dir = os.path.dirname(inst.disks[0].logical_id[1])
3601 result = self.rpc.call_file_storage_dir_rename(inst.primary_node,
3602 old_file_storage_dir,
3603 new_file_storage_dir)
3604 result.Raise("Could not rename on node %s directory '%s' to '%s'"
3605 " (but the instance has been renamed in Ganeti)" %
3606 (inst.primary_node, old_file_storage_dir,
3607 new_file_storage_dir))
3609 _StartInstanceDisks(self, inst, None)
3611 result = self.rpc.call_instance_run_rename(inst.primary_node, inst,
3613 msg = result.fail_msg
3615 msg = ("Could not run OS rename script for instance %s on node %s"
3616 " (but the instance has been renamed in Ganeti): %s" %
3617 (inst.name, inst.primary_node, msg))
3618 self.proc.LogWarning(msg)
3620 _ShutdownInstanceDisks(self, inst)
3623 class LURemoveInstance(LogicalUnit):
3624 """Remove an instance.
3627 HPATH = "instance-remove"
3628 HTYPE = constants.HTYPE_INSTANCE
3629 _OP_REQP = ["instance_name", "ignore_failures"]
3632 def ExpandNames(self):
3633 self._ExpandAndLockInstance()
3634 self.needed_locks[locking.LEVEL_NODE] = []
3635 self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
3637 def DeclareLocks(self, level):
3638 if level == locking.LEVEL_NODE:
3639 self._LockInstancesNodes()
3641 def BuildHooksEnv(self):
3644 This runs on master, primary and secondary nodes of the instance.
3647 env = _BuildInstanceHookEnvByObject(self, self.instance)
3648 nl = [self.cfg.GetMasterNode()]
3651 def CheckPrereq(self):
3652 """Check prerequisites.
3654 This checks that the instance is in the cluster.
3657 self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
3658 assert self.instance is not None, \
3659 "Cannot retrieve locked instance %s" % self.op.instance_name
3661 def Exec(self, feedback_fn):
3662 """Remove the instance.
3665 instance = self.instance
3666 logging.info("Shutting down instance %s on node %s",
3667 instance.name, instance.primary_node)
3669 result = self.rpc.call_instance_shutdown(instance.primary_node, instance)
3670 msg = result.fail_msg
3672 if self.op.ignore_failures:
3673 feedback_fn("Warning: can't shutdown instance: %s" % msg)
3675 raise errors.OpExecError("Could not shutdown instance %s on"
3677 (instance.name, instance.primary_node, msg))
3679 logging.info("Removing block devices for instance %s", instance.name)
3681 if not _RemoveDisks(self, instance):
3682 if self.op.ignore_failures:
3683 feedback_fn("Warning: can't remove instance's disks")
3685 raise errors.OpExecError("Can't remove instance's disks")
3687 logging.info("Removing instance %s out of cluster config", instance.name)
3689 self.cfg.RemoveInstance(instance.name)
3690 self.remove_locks[locking.LEVEL_INSTANCE] = instance.name
3693 class LUQueryInstances(NoHooksLU):
3694 """Logical unit for querying instances.
3697 _OP_REQP = ["output_fields", "names", "use_locking"]
3699 _FIELDS_STATIC = utils.FieldSet(*["name", "os", "pnode", "snodes",
3701 "disk_template", "ip", "mac", "bridge",
3702 "nic_mode", "nic_link",
3703 "sda_size", "sdb_size", "vcpus", "tags",
3704 "network_port", "beparams",
3705 r"(disk)\.(size)/([0-9]+)",
3706 r"(disk)\.(sizes)", "disk_usage",
3707 r"(nic)\.(mac|ip|mode|link)/([0-9]+)",
3708 r"(nic)\.(bridge)/([0-9]+)",
3709 r"(nic)\.(macs|ips|modes|links|bridges)",
3710 r"(disk|nic)\.(count)",
3711 "serial_no", "hypervisor", "hvparams",] +
3713 for name in constants.HVS_PARAMETERS] +
3715 for name in constants.BES_PARAMETERS])
3716 _FIELDS_DYNAMIC = utils.FieldSet("oper_state", "oper_ram", "status")
3719 def ExpandNames(self):
3720 _CheckOutputFields(static=self._FIELDS_STATIC,
3721 dynamic=self._FIELDS_DYNAMIC,
3722 selected=self.op.output_fields)
3724 self.needed_locks = {}
3725 self.share_locks[locking.LEVEL_INSTANCE] = 1
3726 self.share_locks[locking.LEVEL_NODE] = 1
3729 self.wanted = _GetWantedInstances(self, self.op.names)
3731 self.wanted = locking.ALL_SET
3733 self.do_node_query = self._FIELDS_STATIC.NonMatching(self.op.output_fields)
3734 self.do_locking = self.do_node_query and self.op.use_locking
3736 self.needed_locks[locking.LEVEL_INSTANCE] = self.wanted
3737 self.needed_locks[locking.LEVEL_NODE] = []
3738 self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
3740 def DeclareLocks(self, level):
3741 if level == locking.LEVEL_NODE and self.do_locking:
3742 self._LockInstancesNodes()
3744 def CheckPrereq(self):
3745 """Check prerequisites.
3750 def Exec(self, feedback_fn):
3751 """Computes the list of nodes and their attributes.
3754 all_info = self.cfg.GetAllInstancesInfo()
3755 if self.wanted == locking.ALL_SET:
3756 # caller didn't specify instance names, so ordering is not important
3758 instance_names = self.acquired_locks[locking.LEVEL_INSTANCE]
3760 instance_names = all_info.keys()
3761 instance_names = utils.NiceSort(instance_names)
3763 # caller did specify names, so we must keep the ordering
3765 tgt_set = self.acquired_locks[locking.LEVEL_INSTANCE]
3767 tgt_set = all_info.keys()
3768 missing = set(self.wanted).difference(tgt_set)
3770 raise errors.OpExecError("Some instances were removed before"
3771 " retrieving their data: %s" % missing)
3772 instance_names = self.wanted
3774 instance_list = [all_info[iname] for iname in instance_names]
3776 # begin data gathering
3778 nodes = frozenset([inst.primary_node for inst in instance_list])
3779 hv_list = list(set([inst.hypervisor for inst in instance_list]))
3783 if self.do_node_query:
3785 node_data = self.rpc.call_all_instances_info(nodes, hv_list)
3787 result = node_data[name]
3789 # offline nodes will be in both lists
3790 off_nodes.append(name)
3791 if result.failed or result.fail_msg:
3792 bad_nodes.append(name)
3795 live_data.update(result.payload)
3796 # else no instance is alive
3798 live_data = dict([(name, {}) for name in instance_names])
3800 # end data gathering
3805 cluster = self.cfg.GetClusterInfo()
3806 for instance in instance_list:
3808 i_hv = cluster.FillHV(instance)
3809 i_be = cluster.FillBE(instance)
3810 i_nicp = [objects.FillDict(cluster.nicparams[constants.PP_DEFAULT],
3811 nic.nicparams) for nic in instance.nics]
3812 for field in self.op.output_fields:
3813 st_match = self._FIELDS_STATIC.Matches(field)
3818 elif field == "pnode":
3819 val = instance.primary_node
3820 elif field == "snodes":
3821 val = list(instance.secondary_nodes)
3822 elif field == "admin_state":
3823 val = instance.admin_up
3824 elif field == "oper_state":
3825 if instance.primary_node in bad_nodes:
3828 val = bool(live_data.get(instance.name))
3829 elif field == "status":
3830 if instance.primary_node in off_nodes:
3831 val = "ERROR_nodeoffline"
3832 elif instance.primary_node in bad_nodes:
3833 val = "ERROR_nodedown"
3835 running = bool(live_data.get(instance.name))
3837 if instance.admin_up:
3842 if instance.admin_up:
3846 elif field == "oper_ram":
3847 if instance.primary_node in bad_nodes:
3849 elif instance.name in live_data:
3850 val = live_data[instance.name].get("memory", "?")
3853 elif field == "vcpus":
3854 val = i_be[constants.BE_VCPUS]
3855 elif field == "disk_template":
3856 val = instance.disk_template
3859 val = instance.nics[0].ip
3862 elif field == "nic_mode":
3864 val = i_nicp[0][constants.NIC_MODE]
3867 elif field == "nic_link":
3869 val = i_nicp[0][constants.NIC_LINK]
3872 elif field == "bridge":
3873 if (instance.nics and
3874 i_nicp[0][constants.NIC_MODE] == constants.NIC_MODE_BRIDGED):
3875 val = i_nicp[0][constants.NIC_LINK]
3878 elif field == "mac":
3880 val = instance.nics[0].mac
3883 elif field == "sda_size" or field == "sdb_size":
3884 idx = ord(field[2]) - ord('a')
3886 val = instance.FindDisk(idx).size
3887 except errors.OpPrereqError:
3889 elif field == "disk_usage": # total disk usage per node
3890 disk_sizes = [{'size': disk.size} for disk in instance.disks]
3891 val = _ComputeDiskSize(instance.disk_template, disk_sizes)
3892 elif field == "tags":
3893 val = list(instance.GetTags())
3894 elif field == "serial_no":
3895 val = instance.serial_no
3896 elif field == "network_port":
3897 val = instance.network_port
3898 elif field == "hypervisor":
3899 val = instance.hypervisor
3900 elif field == "hvparams":
3902 elif (field.startswith(HVPREFIX) and
3903 field[len(HVPREFIX):] in constants.HVS_PARAMETERS):
3904 val = i_hv.get(field[len(HVPREFIX):], None)
3905 elif field == "beparams":
3907 elif (field.startswith(BEPREFIX) and
3908 field[len(BEPREFIX):] in constants.BES_PARAMETERS):
3909 val = i_be.get(field[len(BEPREFIX):], None)
3910 elif st_match and st_match.groups():
3911 # matches a variable list
3912 st_groups = st_match.groups()
3913 if st_groups and st_groups[0] == "disk":
3914 if st_groups[1] == "count":
3915 val = len(instance.disks)
3916 elif st_groups[1] == "sizes":
3917 val = [disk.size for disk in instance.disks]
3918 elif st_groups[1] == "size":
3920 val = instance.FindDisk(st_groups[2]).size
3921 except errors.OpPrereqError:
3924 assert False, "Unhandled disk parameter"
3925 elif st_groups[0] == "nic":
3926 if st_groups[1] == "count":
3927 val = len(instance.nics)
3928 elif st_groups[1] == "macs":
3929 val = [nic.mac for nic in instance.nics]
3930 elif st_groups[1] == "ips":
3931 val = [nic.ip for nic in instance.nics]
3932 elif st_groups[1] == "modes":
3933 val = [nicp[constants.NIC_MODE] for nicp in i_nicp]
3934 elif st_groups[1] == "links":
3935 val = [nicp[constants.NIC_LINK] for nicp in i_nicp]
3936 elif st_groups[1] == "bridges":
3939 if nicp[constants.NIC_MODE] == constants.NIC_MODE_BRIDGED:
3940 val.append(nicp[constants.NIC_LINK])
3945 nic_idx = int(st_groups[2])
3946 if nic_idx >= len(instance.nics):
3949 if st_groups[1] == "mac":
3950 val = instance.nics[nic_idx].mac
3951 elif st_groups[1] == "ip":
3952 val = instance.nics[nic_idx].ip
3953 elif st_groups[1] == "mode":
3954 val = i_nicp[nic_idx][constants.NIC_MODE]
3955 elif st_groups[1] == "link":
3956 val = i_nicp[nic_idx][constants.NIC_LINK]
3957 elif st_groups[1] == "bridge":
3958 nic_mode = i_nicp[nic_idx][constants.NIC_MODE]
3959 if nic_mode == constants.NIC_MODE_BRIDGED:
3960 val = i_nicp[nic_idx][constants.NIC_LINK]
3964 assert False, "Unhandled NIC parameter"
3966 assert False, ("Declared but unhandled variable parameter '%s'" %
3969 assert False, "Declared but unhandled parameter '%s'" % field
3976 class LUFailoverInstance(LogicalUnit):
3977 """Failover an instance.
3980 HPATH = "instance-failover"
3981 HTYPE = constants.HTYPE_INSTANCE
3982 _OP_REQP = ["instance_name", "ignore_consistency"]
3985 def ExpandNames(self):
3986 self._ExpandAndLockInstance()
3987 self.needed_locks[locking.LEVEL_NODE] = []
3988 self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
3990 def DeclareLocks(self, level):
3991 if level == locking.LEVEL_NODE:
3992 self._LockInstancesNodes()
3994 def BuildHooksEnv(self):
3997 This runs on master, primary and secondary nodes of the instance.
4001 "IGNORE_CONSISTENCY": self.op.ignore_consistency,
4003 env.update(_BuildInstanceHookEnvByObject(self, self.instance))
4004 nl = [self.cfg.GetMasterNode()] + list(self.instance.secondary_nodes)
4007 def CheckPrereq(self):
4008 """Check prerequisites.
4010 This checks that the instance is in the cluster.
4013 self.instance = instance = self.cfg.GetInstanceInfo(self.op.instance_name)
4014 assert self.instance is not None, \
4015 "Cannot retrieve locked instance %s" % self.op.instance_name
4017 bep = self.cfg.GetClusterInfo().FillBE(instance)
4018 if instance.disk_template not in constants.DTS_NET_MIRROR:
4019 raise errors.OpPrereqError("Instance's disk layout is not"
4020 " network mirrored, cannot failover.")
4022 secondary_nodes = instance.secondary_nodes
4023 if not secondary_nodes:
4024 raise errors.ProgrammerError("no secondary node but using "
4025 "a mirrored disk template")
4027 target_node = secondary_nodes[0]
4028 _CheckNodeOnline(self, target_node)
4029 _CheckNodeNotDrained(self, target_node)
4030 if instance.admin_up:
4031 # check memory requirements on the secondary node
4032 _CheckNodeFreeMemory(self, target_node, "failing over instance %s" %
4033 instance.name, bep[constants.BE_MEMORY],
4034 instance.hypervisor)
4036 self.LogInfo("Not checking memory on the secondary node as"
4037 " instance will not be started")
4039 # check bridge existance
4040 _CheckInstanceBridgesExist(self, instance, node=target_node)
4042 def Exec(self, feedback_fn):
4043 """Failover an instance.
4045 The failover is done by shutting it down on its present node and
4046 starting it on the secondary.
4049 instance = self.instance
4051 source_node = instance.primary_node
4052 target_node = instance.secondary_nodes[0]
4054 feedback_fn("* checking disk consistency between source and target")
4055 for dev in instance.disks:
4056 # for drbd, these are drbd over lvm
4057 if not _CheckDiskConsistency(self, dev, target_node, False):
4058 if instance.admin_up and not self.op.ignore_consistency:
4059 raise errors.OpExecError("Disk %s is degraded on target node,"
4060 " aborting failover." % dev.iv_name)
4062 feedback_fn("* shutting down instance on source node")
4063 logging.info("Shutting down instance %s on node %s",
4064 instance.name, source_node)
4066 result = self.rpc.call_instance_shutdown(source_node, instance)
4067 msg = result.fail_msg
4069 if self.op.ignore_consistency:
4070 self.proc.LogWarning("Could not shutdown instance %s on node %s."
4071 " Proceeding anyway. Please make sure node"
4072 " %s is down. Error details: %s",
4073 instance.name, source_node, source_node, msg)
4075 raise errors.OpExecError("Could not shutdown instance %s on"
4077 (instance.name, source_node, msg))
4079 feedback_fn("* deactivating the instance's disks on source node")
4080 if not _ShutdownInstanceDisks(self, instance, ignore_primary=True):
4081 raise errors.OpExecError("Can't shut down the instance's disks.")
4083 instance.primary_node = target_node
4084 # distribute new instance config to the other nodes
4085 self.cfg.Update(instance)
4087 # Only start the instance if it's marked as up
4088 if instance.admin_up:
4089 feedback_fn("* activating the instance's disks on target node")
4090 logging.info("Starting instance %s on node %s",
4091 instance.name, target_node)
4093 disks_ok, _ = _AssembleInstanceDisks(self, instance,
4094 ignore_secondaries=True)
4096 _ShutdownInstanceDisks(self, instance)
4097 raise errors.OpExecError("Can't activate the instance's disks")
4099 feedback_fn("* starting the instance on the target node")
4100 result = self.rpc.call_instance_start(target_node, instance, None, None)
4101 msg = result.fail_msg
4103 _ShutdownInstanceDisks(self, instance)
4104 raise errors.OpExecError("Could not start instance %s on node %s: %s" %
4105 (instance.name, target_node, msg))
4108 class LUMigrateInstance(LogicalUnit):
4109 """Migrate an instance.
4111 This is migration without shutting down, compared to the failover,
4112 which is done with shutdown.
4115 HPATH = "instance-migrate"
4116 HTYPE = constants.HTYPE_INSTANCE
4117 _OP_REQP = ["instance_name", "live", "cleanup"]
4121 def ExpandNames(self):
4122 self._ExpandAndLockInstance()
4124 self.needed_locks[locking.LEVEL_NODE] = []
4125 self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
4127 self._migrater = TLMigrateInstance(self, self.op.instance_name,
4128 self.op.live, self.op.cleanup)
4129 self.tasklets = [self._migrater]
4131 def DeclareLocks(self, level):
4132 if level == locking.LEVEL_NODE:
4133 self._LockInstancesNodes()
4135 def BuildHooksEnv(self):
4138 This runs on master, primary and secondary nodes of the instance.
4141 instance = self._migrater.instance
4142 env = _BuildInstanceHookEnvByObject(self, instance)
4143 env["MIGRATE_LIVE"] = self.op.live
4144 env["MIGRATE_CLEANUP"] = self.op.cleanup
4145 nl = [self.cfg.GetMasterNode()] + list(instance.secondary_nodes)
4149 class LUMigrateNode(LogicalUnit):
4150 """Migrate all instances from a node.
4153 HPATH = "node-migrate"
4154 HTYPE = constants.HTYPE_NODE
4155 _OP_REQP = ["node_name", "live"]
4158 def ExpandNames(self):
4159 self.op.node_name = self.cfg.ExpandNodeName(self.op.node_name)
4160 if self.op.node_name is None:
4161 raise errors.OpPrereqError("Node '%s' not known" % self.op.node_name)
4163 self.needed_locks = {
4164 locking.LEVEL_NODE: [self.op.node_name],
4167 self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_APPEND
4169 # Create tasklets for migrating instances for all instances on this node
4173 for inst in _GetNodePrimaryInstances(self.cfg, self.op.node_name):
4174 logging.debug("Migrating instance %s", inst.name)
4175 names.append(inst.name)
4177 tasklets.append(TLMigrateInstance(self, inst.name, self.op.live, False))
4179 self.tasklets = tasklets
4181 # Declare instance locks
4182 self.needed_locks[locking.LEVEL_INSTANCE] = names
4184 def DeclareLocks(self, level):
4185 if level == locking.LEVEL_NODE:
4186 self._LockInstancesNodes()
4188 def BuildHooksEnv(self):
4191 This runs on the master, the primary and all the secondaries.
4195 "NODE_NAME": self.op.node_name,
4198 nl = [self.cfg.GetMasterNode()]
4200 return (env, nl, nl)
4203 class TLMigrateInstance(Tasklet):
4204 def __init__(self, lu, instance_name, live, cleanup):
4205 """Initializes this class.
4208 Tasklet.__init__(self, lu)
4211 self.instance_name = instance_name
4213 self.cleanup = cleanup
4215 def CheckPrereq(self):
4216 """Check prerequisites.
4218 This checks that the instance is in the cluster.
4221 instance = self.cfg.GetInstanceInfo(
4222 self.cfg.ExpandInstanceName(self.instance_name))
4223 if instance is None:
4224 raise errors.OpPrereqError("Instance '%s' not known" %
4227 if instance.disk_template != constants.DT_DRBD8:
4228 raise errors.OpPrereqError("Instance's disk layout is not"
4229 " drbd8, cannot migrate.")
4231 secondary_nodes = instance.secondary_nodes
4232 if not secondary_nodes:
4233 raise errors.ConfigurationError("No secondary node but using"
4234 " drbd8 disk template")
4236 i_be = self.cfg.GetClusterInfo().FillBE(instance)
4238 target_node = secondary_nodes[0]
4239 # check memory requirements on the secondary node
4240 _CheckNodeFreeMemory(self, target_node, "migrating instance %s" %
4241 instance.name, i_be[constants.BE_MEMORY],
4242 instance.hypervisor)
4244 # check bridge existance
4245 _CheckInstanceBridgesExist(self, instance, node=target_node)
4247 if not self.cleanup:
4248 _CheckNodeNotDrained(self, target_node)
4249 result = self.rpc.call_instance_migratable(instance.primary_node,
4251 result.Raise("Can't migrate, please use failover", prereq=True)
4253 self.instance = instance
4255 def _WaitUntilSync(self):
4256 """Poll with custom rpc for disk sync.
4258 This uses our own step-based rpc call.
4261 self.feedback_fn("* wait until resync is done")
4265 result = self.rpc.call_drbd_wait_sync(self.all_nodes,
4267 self.instance.disks)
4269 for node, nres in result.items():
4270 nres.Raise("Cannot resync disks on node %s" % node)
4271 node_done, node_percent = nres.payload
4272 all_done = all_done and node_done
4273 if node_percent is not None:
4274 min_percent = min(min_percent, node_percent)
4276 if min_percent < 100:
4277 self.feedback_fn(" - progress: %.1f%%" % min_percent)
4280 def _EnsureSecondary(self, node):
4281 """Demote a node to secondary.
4284 self.feedback_fn("* switching node %s to secondary mode" % node)
4286 for dev in self.instance.disks:
4287 self.cfg.SetDiskID(dev, node)
4289 result = self.rpc.call_blockdev_close(node, self.instance.name,
4290 self.instance.disks)
4291 result.Raise("Cannot change disk to secondary on node %s" % node)
4293 def _GoStandalone(self):
4294 """Disconnect from the network.
4297 self.feedback_fn("* changing into standalone mode")
4298 result = self.rpc.call_drbd_disconnect_net(self.all_nodes, self.nodes_ip,
4299 self.instance.disks)
4300 for node, nres in result.items():
4301 nres.Raise("Cannot disconnect disks node %s" % node)
4303 def _GoReconnect(self, multimaster):
4304 """Reconnect to the network.
4310 msg = "single-master"
4311 self.feedback_fn("* changing disks into %s mode" % msg)
4312 result = self.rpc.call_drbd_attach_net(self.all_nodes, self.nodes_ip,
4313 self.instance.disks,
4314 self.instance.name, multimaster)
4315 for node, nres in result.items():
4316 nres.Raise("Cannot change disks config on node %s" % node)
4318 def _ExecCleanup(self):
4319 """Try to cleanup after a failed migration.
4321 The cleanup is done by:
4322 - check that the instance is running only on one node
4323 (and update the config if needed)
4324 - change disks on its secondary node to secondary
4325 - wait until disks are fully synchronized
4326 - disconnect from the network
4327 - change disks into single-master mode
4328 - wait again until disks are fully synchronized
4331 instance = self.instance
4332 target_node = self.target_node
4333 source_node = self.source_node
4335 # check running on only one node
4336 self.feedback_fn("* checking where the instance actually runs"
4337 " (if this hangs, the hypervisor might be in"
4339 ins_l = self.rpc.call_instance_list(self.all_nodes, [instance.hypervisor])
4340 for node, result in ins_l.items():
4341 result.Raise("Can't contact node %s" % node)
4343 runningon_source = instance.name in ins_l[source_node].payload
4344 runningon_target = instance.name in ins_l[target_node].payload
4346 if runningon_source and runningon_target:
4347 raise errors.OpExecError("Instance seems to be running on two nodes,"
4348 " or the hypervisor is confused. You will have"
4349 " to ensure manually that it runs only on one"
4350 " and restart this operation.")
4352 if not (runningon_source or runningon_target):
4353 raise errors.OpExecError("Instance does not seem to be running at all."
4354 " In this case, it's safer to repair by"
4355 " running 'gnt-instance stop' to ensure disk"
4356 " shutdown, and then restarting it.")
4358 if runningon_target:
4359 # the migration has actually succeeded, we need to update the config
4360 self.feedback_fn("* instance running on secondary node (%s),"
4361 " updating config" % target_node)
4362 instance.primary_node = target_node
4363 self.cfg.Update(instance)
4364 demoted_node = source_node
4366 self.feedback_fn("* instance confirmed to be running on its"
4367 " primary node (%s)" % source_node)
4368 demoted_node = target_node
4370 self._EnsureSecondary(demoted_node)
4372 self._WaitUntilSync()
4373 except errors.OpExecError:
4374 # we ignore here errors, since if the device is standalone, it
4375 # won't be able to sync
4377 self._GoStandalone()
4378 self._GoReconnect(False)
4379 self._WaitUntilSync()
4381 self.feedback_fn("* done")
4383 def _RevertDiskStatus(self):
4384 """Try to revert the disk status after a failed migration.
4387 target_node = self.target_node
4389 self._EnsureSecondary(target_node)
4390 self._GoStandalone()
4391 self._GoReconnect(False)
4392 self._WaitUntilSync()
4393 except errors.OpExecError, err:
4394 self.lu.LogWarning("Migration failed and I can't reconnect the"
4395 " drives: error '%s'\n"
4396 "Please look and recover the instance status" %
4399 def _AbortMigration(self):
4400 """Call the hypervisor code to abort a started migration.
4403 instance = self.instance
4404 target_node = self.target_node
4405 migration_info = self.migration_info
4407 abort_result = self.rpc.call_finalize_migration(target_node,
4411 abort_msg = abort_result.fail_msg
4413 logging.error("Aborting migration failed on target node %s: %s" %
4414 (target_node, abort_msg))
4415 # Don't raise an exception here, as we stil have to try to revert the
4416 # disk status, even if this step failed.
4418 def _ExecMigration(self):
4419 """Migrate an instance.
4421 The migrate is done by:
4422 - change the disks into dual-master mode
4423 - wait until disks are fully synchronized again
4424 - migrate the instance
4425 - change disks on the new secondary node (the old primary) to secondary
4426 - wait until disks are fully synchronized
4427 - change disks into single-master mode
4430 instance = self.instance
4431 target_node = self.target_node
4432 source_node = self.source_node
4434 self.feedback_fn("* checking disk consistency between source and target")
4435 for dev in instance.disks:
4436 if not _CheckDiskConsistency(self, dev, target_node, False):
4437 raise errors.OpExecError("Disk %s is degraded or not fully"
4438 " synchronized on target node,"
4439 " aborting migrate." % dev.iv_name)
4441 # First get the migration information from the remote node
4442 result = self.rpc.call_migration_info(source_node, instance)
4443 msg = result.fail_msg
4445 log_err = ("Failed fetching source migration information from %s: %s" %
4447 logging.error(log_err)
4448 raise errors.OpExecError(log_err)
4450 self.migration_info = migration_info = result.payload
4452 # Then switch the disks to master/master mode
4453 self._EnsureSecondary(target_node)
4454 self._GoStandalone()
4455 self._GoReconnect(True)
4456 self._WaitUntilSync()
4458 self.feedback_fn("* preparing %s to accept the instance" % target_node)
4459 result = self.rpc.call_accept_instance(target_node,
4462 self.nodes_ip[target_node])
4464 msg = result.fail_msg
4466 logging.error("Instance pre-migration failed, trying to revert"
4467 " disk status: %s", msg)
4468 self._AbortMigration()
4469 self._RevertDiskStatus()
4470 raise errors.OpExecError("Could not pre-migrate instance %s: %s" %
4471 (instance.name, msg))
4473 self.feedback_fn("* migrating instance to %s" % target_node)
4475 result = self.rpc.call_instance_migrate(source_node, instance,
4476 self.nodes_ip[target_node],
4478 msg = result.fail_msg
4480 logging.error("Instance migration failed, trying to revert"
4481 " disk status: %s", msg)
4482 self._AbortMigration()
4483 self._RevertDiskStatus()
4484 raise errors.OpExecError("Could not migrate instance %s: %s" %
4485 (instance.name, msg))
4488 instance.primary_node = target_node
4489 # distribute new instance config to the other nodes
4490 self.cfg.Update(instance)
4492 result = self.rpc.call_finalize_migration(target_node,
4496 msg = result.fail_msg
4498 logging.error("Instance migration succeeded, but finalization failed:"
4500 raise errors.OpExecError("Could not finalize instance migration: %s" %
4503 self._EnsureSecondary(source_node)
4504 self._WaitUntilSync()
4505 self._GoStandalone()
4506 self._GoReconnect(False)
4507 self._WaitUntilSync()
4509 self.feedback_fn("* done")
4511 def Exec(self, feedback_fn):
4512 """Perform the migration.
4515 feedback_fn("Migrating instance %s" % self.instance.name)
4517 self.feedback_fn = feedback_fn
4519 self.source_node = self.instance.primary_node
4520 self.target_node = self.instance.secondary_nodes[0]
4521 self.all_nodes = [self.source_node, self.target_node]
4523 self.source_node: self.cfg.GetNodeInfo(self.source_node).secondary_ip,
4524 self.target_node: self.cfg.GetNodeInfo(self.target_node).secondary_ip,
4528 return self._ExecCleanup()
4530 return self._ExecMigration()
4533 def _CreateBlockDev(lu, node, instance, device, force_create,
4535 """Create a tree of block devices on a given node.
4537 If this device type has to be created on secondaries, create it and
4540 If not, just recurse to children keeping the same 'force' value.
4542 @param lu: the lu on whose behalf we execute
4543 @param node: the node on which to create the device
4544 @type instance: L{objects.Instance}
4545 @param instance: the instance which owns the device
4546 @type device: L{objects.Disk}
4547 @param device: the device to create
4548 @type force_create: boolean
4549 @param force_create: whether to force creation of this device; this
4550 will be change to True whenever we find a device which has
4551 CreateOnSecondary() attribute
4552 @param info: the extra 'metadata' we should attach to the device
4553 (this will be represented as a LVM tag)
4554 @type force_open: boolean
4555 @param force_open: this parameter will be passes to the
4556 L{backend.BlockdevCreate} function where it specifies
4557 whether we run on primary or not, and it affects both
4558 the child assembly and the device own Open() execution
4561 if device.CreateOnSecondary():
4565 for child in device.children:
4566 _CreateBlockDev(lu, node, instance, child, force_create,
4569 if not force_create:
4572 _CreateSingleBlockDev(lu, node, instance, device, info, force_open)
4575 def _CreateSingleBlockDev(lu, node, instance, device, info, force_open):
4576 """Create a single block device on a given node.
4578 This will not recurse over children of the device, so they must be
4581 @param lu: the lu on whose behalf we execute
4582 @param node: the node on which to create the device
4583 @type instance: L{objects.Instance}
4584 @param instance: the instance which owns the device
4585 @type device: L{objects.Disk}
4586 @param device: the device to create
4587 @param info: the extra 'metadata' we should attach to the device
4588 (this will be represented as a LVM tag)
4589 @type force_open: boolean
4590 @param force_open: this parameter will be passes to the
4591 L{backend.BlockdevCreate} function where it specifies
4592 whether we run on primary or not, and it affects both
4593 the child assembly and the device own Open() execution
4596 lu.cfg.SetDiskID(device, node)
4597 result = lu.rpc.call_blockdev_create(node, device, device.size,
4598 instance.name, force_open, info)
4599 result.Raise("Can't create block device %s on"
4600 " node %s for instance %s" % (device, node, instance.name))
4601 if device.physical_id is None:
4602 device.physical_id = result.payload
4605 def _GenerateUniqueNames(lu, exts):
4606 """Generate a suitable LV name.
4608 This will generate a logical volume name for the given instance.
4613 new_id = lu.cfg.GenerateUniqueID()
4614 results.append("%s%s" % (new_id, val))
4618 def _GenerateDRBD8Branch(lu, primary, secondary, size, names, iv_name,
4620 """Generate a drbd8 device complete with its children.
4623 port = lu.cfg.AllocatePort()
4624 vgname = lu.cfg.GetVGName()
4625 shared_secret = lu.cfg.GenerateDRBDSecret()
4626 dev_data = objects.Disk(dev_type=constants.LD_LV, size=size,
4627 logical_id=(vgname, names[0]))
4628 dev_meta = objects.Disk(dev_type=constants.LD_LV, size=128,
4629 logical_id=(vgname, names[1]))
4630 drbd_dev = objects.Disk(dev_type=constants.LD_DRBD8, size=size,
4631 logical_id=(primary, secondary, port,
4634 children=[dev_data, dev_meta],
4639 def _GenerateDiskTemplate(lu, template_name,
4640 instance_name, primary_node,
4641 secondary_nodes, disk_info,
4642 file_storage_dir, file_driver,
4644 """Generate the entire disk layout for a given template type.
4647 #TODO: compute space requirements
4649 vgname = lu.cfg.GetVGName()
4650 disk_count = len(disk_info)
4652 if template_name == constants.DT_DISKLESS:
4654 elif template_name == constants.DT_PLAIN:
4655 if len(secondary_nodes) != 0:
4656 raise errors.ProgrammerError("Wrong template configuration")
4658 names = _GenerateUniqueNames(lu, [".disk%d" % (base_index + i)
4659 for i in range(disk_count)])
4660 for idx, disk in enumerate(disk_info):
4661 disk_index = idx + base_index
4662 disk_dev = objects.Disk(dev_type=constants.LD_LV, size=disk["size"],
4663 logical_id=(vgname, names[idx]),
4664 iv_name="disk/%d" % disk_index,
4666 disks.append(disk_dev)
4667 elif template_name == constants.DT_DRBD8:
4668 if len(secondary_nodes) != 1:
4669 raise errors.ProgrammerError("Wrong template configuration")
4670 remote_node = secondary_nodes[0]
4671 minors = lu.cfg.AllocateDRBDMinor(
4672 [primary_node, remote_node] * len(disk_info), instance_name)
4675 for lv_prefix in _GenerateUniqueNames(lu, [".disk%d" % (base_index + i)
4676 for i in range(disk_count)]):
4677 names.append(lv_prefix + "_data")
4678 names.append(lv_prefix + "_meta")
4679 for idx, disk in enumerate(disk_info):
4680 disk_index = idx + base_index
4681 disk_dev = _GenerateDRBD8Branch(lu, primary_node, remote_node,
4682 disk["size"], names[idx*2:idx*2+2],
4683 "disk/%d" % disk_index,
4684 minors[idx*2], minors[idx*2+1])
4685 disk_dev.mode = disk["mode"]
4686 disks.append(disk_dev)
4687 elif template_name == constants.DT_FILE:
4688 if len(secondary_nodes) != 0:
4689 raise errors.ProgrammerError("Wrong template configuration")
4691 for idx, disk in enumerate(disk_info):
4692 disk_index = idx + base_index
4693 disk_dev = objects.Disk(dev_type=constants.LD_FILE, size=disk["size"],
4694 iv_name="disk/%d" % disk_index,
4695 logical_id=(file_driver,
4696 "%s/disk%d" % (file_storage_dir,
4699 disks.append(disk_dev)
4701 raise errors.ProgrammerError("Invalid disk template '%s'" % template_name)
4705 def _GetInstanceInfoText(instance):
4706 """Compute that text that should be added to the disk's metadata.
4709 return "originstname+%s" % instance.name
4712 def _CreateDisks(lu, instance):
4713 """Create all disks for an instance.
4715 This abstracts away some work from AddInstance.
4717 @type lu: L{LogicalUnit}
4718 @param lu: the logical unit on whose behalf we execute
4719 @type instance: L{objects.Instance}
4720 @param instance: the instance whose disks we should create
4722 @return: the success of the creation
4725 info = _GetInstanceInfoText(instance)
4726 pnode = instance.primary_node
4728 if instance.disk_template == constants.DT_FILE:
4729 file_storage_dir = os.path.dirname(instance.disks[0].logical_id[1])
4730 result = lu.rpc.call_file_storage_dir_create(pnode, file_storage_dir)
4732 result.Raise("Failed to create directory '%s' on"
4733 " node %s: %s" % (file_storage_dir, pnode))
4735 # Note: this needs to be kept in sync with adding of disks in
4736 # LUSetInstanceParams
4737 for device in instance.disks:
4738 logging.info("Creating volume %s for instance %s",
4739 device.iv_name, instance.name)
4741 for node in instance.all_nodes:
4742 f_create = node == pnode
4743 _CreateBlockDev(lu, node, instance, device, f_create, info, f_create)
4746 def _RemoveDisks(lu, instance):
4747 """Remove all disks for an instance.
4749 This abstracts away some work from `AddInstance()` and
4750 `RemoveInstance()`. Note that in case some of the devices couldn't
4751 be removed, the removal will continue with the other ones (compare
4752 with `_CreateDisks()`).
4754 @type lu: L{LogicalUnit}
4755 @param lu: the logical unit on whose behalf we execute
4756 @type instance: L{objects.Instance}
4757 @param instance: the instance whose disks we should remove
4759 @return: the success of the removal
4762 logging.info("Removing block devices for instance %s", instance.name)
4765 for device in instance.disks:
4766 for node, disk in device.ComputeNodeTree(instance.primary_node):
4767 lu.cfg.SetDiskID(disk, node)
4768 msg = lu.rpc.call_blockdev_remove(node, disk).fail_msg
4770 lu.LogWarning("Could not remove block device %s on node %s,"
4771 " continuing anyway: %s", device.iv_name, node, msg)
4774 if instance.disk_template == constants.DT_FILE:
4775 file_storage_dir = os.path.dirname(instance.disks[0].logical_id[1])
4776 result = lu.rpc.call_file_storage_dir_remove(instance.primary_node,
4778 msg = result.fail_msg
4780 lu.LogWarning("Could not remove directory '%s' on node %s: %s",
4781 file_storage_dir, instance.primary_node, msg)
4787 def _ComputeDiskSize(disk_template, disks):
4788 """Compute disk size requirements in the volume group
4791 # Required free disk space as a function of disk and swap space
4793 constants.DT_DISKLESS: None,
4794 constants.DT_PLAIN: sum(d["size"] for d in disks),
4795 # 128 MB are added for drbd metadata for each disk
4796 constants.DT_DRBD8: sum(d["size"] + 128 for d in disks),
4797 constants.DT_FILE: None,
4800 if disk_template not in req_size_dict:
4801 raise errors.ProgrammerError("Disk template '%s' size requirement"
4802 " is unknown" % disk_template)
4804 return req_size_dict[disk_template]
4807 def _CheckHVParams(lu, nodenames, hvname, hvparams):
4808 """Hypervisor parameter validation.
4810 This function abstract the hypervisor parameter validation to be
4811 used in both instance create and instance modify.
4813 @type lu: L{LogicalUnit}
4814 @param lu: the logical unit for which we check
4815 @type nodenames: list
4816 @param nodenames: the list of nodes on which we should check
4817 @type hvname: string
4818 @param hvname: the name of the hypervisor we should use
4819 @type hvparams: dict
4820 @param hvparams: the parameters which we need to check
4821 @raise errors.OpPrereqError: if the parameters are not valid
4824 hvinfo = lu.rpc.call_hypervisor_validate_params(nodenames,
4827 for node in nodenames:
4831 info.Raise("Hypervisor parameter validation failed on node %s" % node)
4834 class LUCreateInstance(LogicalUnit):
4835 """Create an instance.
4838 HPATH = "instance-add"
4839 HTYPE = constants.HTYPE_INSTANCE
4840 _OP_REQP = ["instance_name", "disks", "disk_template",
4842 "wait_for_sync", "ip_check", "nics",
4843 "hvparams", "beparams"]
4846 def _ExpandNode(self, node):
4847 """Expands and checks one node name.
4850 node_full = self.cfg.ExpandNodeName(node)
4851 if node_full is None:
4852 raise errors.OpPrereqError("Unknown node %s" % node)
4855 def ExpandNames(self):
4856 """ExpandNames for CreateInstance.
4858 Figure out the right locks for instance creation.
4861 self.needed_locks = {}
4863 # set optional parameters to none if they don't exist
4864 for attr in ["pnode", "snode", "iallocator", "hypervisor"]:
4865 if not hasattr(self.op, attr):
4866 setattr(self.op, attr, None)
4868 # cheap checks, mostly valid constants given
4870 # verify creation mode
4871 if self.op.mode not in (constants.INSTANCE_CREATE,
4872 constants.INSTANCE_IMPORT):
4873 raise errors.OpPrereqError("Invalid instance creation mode '%s'" %
4876 # disk template and mirror node verification
4877 if self.op.disk_template not in constants.DISK_TEMPLATES:
4878 raise errors.OpPrereqError("Invalid disk template name")
4880 if self.op.hypervisor is None:
4881 self.op.hypervisor = self.cfg.GetHypervisorType()
4883 cluster = self.cfg.GetClusterInfo()
4884 enabled_hvs = cluster.enabled_hypervisors
4885 if self.op.hypervisor not in enabled_hvs:
4886 raise errors.OpPrereqError("Selected hypervisor (%s) not enabled in the"
4887 " cluster (%s)" % (self.op.hypervisor,
4888 ",".join(enabled_hvs)))
4890 # check hypervisor parameter syntax (locally)
4891 utils.ForceDictType(self.op.hvparams, constants.HVS_PARAMETER_TYPES)
4892 filled_hvp = objects.FillDict(cluster.hvparams[self.op.hypervisor],
4894 hv_type = hypervisor.GetHypervisor(self.op.hypervisor)
4895 hv_type.CheckParameterSyntax(filled_hvp)
4896 self.hv_full = filled_hvp
4898 # fill and remember the beparams dict
4899 utils.ForceDictType(self.op.beparams, constants.BES_PARAMETER_TYPES)
4900 self.be_full = objects.FillDict(cluster.beparams[constants.PP_DEFAULT],
4903 #### instance parameters check
4905 # instance name verification
4906 hostname1 = utils.HostInfo(self.op.instance_name)
4907 self.op.instance_name = instance_name = hostname1.name
4909 # this is just a preventive check, but someone might still add this
4910 # instance in the meantime, and creation will fail at lock-add time
4911 if instance_name in self.cfg.GetInstanceList():
4912 raise errors.OpPrereqError("Instance '%s' is already in the cluster" %
4915 self.add_locks[locking.LEVEL_INSTANCE] = instance_name
4919 for idx, nic in enumerate(self.op.nics):
4920 nic_mode_req = nic.get("mode", None)
4921 nic_mode = nic_mode_req
4922 if nic_mode is None:
4923 nic_mode = cluster.nicparams[constants.PP_DEFAULT][constants.NIC_MODE]
4925 # in routed mode, for the first nic, the default ip is 'auto'
4926 if nic_mode == constants.NIC_MODE_ROUTED and idx == 0:
4927 default_ip_mode = constants.VALUE_AUTO
4929 default_ip_mode = constants.VALUE_NONE
4931 # ip validity checks
4932 ip = nic.get("ip", default_ip_mode)
4933 if ip is None or ip.lower() == constants.VALUE_NONE:
4935 elif ip.lower() == constants.VALUE_AUTO:
4936 nic_ip = hostname1.ip
4938 if not utils.IsValidIP(ip):
4939 raise errors.OpPrereqError("Given IP address '%s' doesn't look"
4940 " like a valid IP" % ip)
4943 # TODO: check the ip for uniqueness !!
4944 if nic_mode == constants.NIC_MODE_ROUTED and not nic_ip:
4945 raise errors.OpPrereqError("Routed nic mode requires an ip address")
4947 # MAC address verification
4948 mac = nic.get("mac", constants.VALUE_AUTO)
4949 if mac not in (constants.VALUE_AUTO, constants.VALUE_GENERATE):
4950 if not utils.IsValidMac(mac.lower()):
4951 raise errors.OpPrereqError("Invalid MAC address specified: %s" %
4953 # bridge verification
4954 bridge = nic.get("bridge", None)
4955 link = nic.get("link", None)
4957 raise errors.OpPrereqError("Cannot pass 'bridge' and 'link'"
4958 " at the same time")
4959 elif bridge and nic_mode == constants.NIC_MODE_ROUTED:
4960 raise errors.OpPrereqError("Cannot pass 'bridge' on a routed nic")
4966 nicparams[constants.NIC_MODE] = nic_mode_req
4968 nicparams[constants.NIC_LINK] = link
4970 check_params = objects.FillDict(cluster.nicparams[constants.PP_DEFAULT],
4972 objects.NIC.CheckParameterSyntax(check_params)
4973 self.nics.append(objects.NIC(mac=mac, ip=nic_ip, nicparams=nicparams))
4975 # disk checks/pre-build
4977 for disk in self.op.disks:
4978 mode = disk.get("mode", constants.DISK_RDWR)
4979 if mode not in constants.DISK_ACCESS_SET:
4980 raise errors.OpPrereqError("Invalid disk access mode '%s'" %
4982 size = disk.get("size", None)
4984 raise errors.OpPrereqError("Missing disk size")
4988 raise errors.OpPrereqError("Invalid disk size '%s'" % size)
4989 self.disks.append({"size": size, "mode": mode})
4991 # used in CheckPrereq for ip ping check
4992 self.check_ip = hostname1.ip
4994 # file storage checks
4995 if (self.op.file_driver and
4996 not self.op.file_driver in constants.FILE_DRIVER):
4997 raise errors.OpPrereqError("Invalid file driver name '%s'" %
4998 self.op.file_driver)
5000 if self.op.file_storage_dir and os.path.isabs(self.op.file_storage_dir):
5001 raise errors.OpPrereqError("File storage directory path not absolute")
5003 ### Node/iallocator related checks
5004 if [self.op.iallocator, self.op.pnode].count(None) != 1:
5005 raise errors.OpPrereqError("One and only one of iallocator and primary"
5006 " node must be given")
5008 if self.op.iallocator:
5009 self.needed_locks[locking.LEVEL_NODE] = locking.ALL_SET
5011 self.op.pnode = self._ExpandNode(self.op.pnode)
5012 nodelist = [self.op.pnode]
5013 if self.op.snode is not None:
5014 self.op.snode = self._ExpandNode(self.op.snode)
5015 nodelist.append(self.op.snode)
5016 self.needed_locks[locking.LEVEL_NODE] = nodelist
5018 # in case of import lock the source node too
5019 if self.op.mode == constants.INSTANCE_IMPORT:
5020 src_node = getattr(self.op, "src_node", None)
5021 src_path = getattr(self.op, "src_path", None)
5023 if src_path is None:
5024 self.op.src_path = src_path = self.op.instance_name
5026 if src_node is None:
5027 self.needed_locks[locking.LEVEL_NODE] = locking.ALL_SET
5028 self.op.src_node = None
5029 if os.path.isabs(src_path):
5030 raise errors.OpPrereqError("Importing an instance from an absolute"
5031 " path requires a source node option.")
5033 self.op.src_node = src_node = self._ExpandNode(src_node)
5034 if self.needed_locks[locking.LEVEL_NODE] is not locking.ALL_SET:
5035 self.needed_locks[locking.LEVEL_NODE].append(src_node)
5036 if not os.path.isabs(src_path):
5037 self.op.src_path = src_path = \
5038 os.path.join(constants.EXPORT_DIR, src_path)
5040 else: # INSTANCE_CREATE
5041 if getattr(self.op, "os_type", None) is None:
5042 raise errors.OpPrereqError("No guest OS specified")
5044 def _RunAllocator(self):
5045 """Run the allocator based on input opcode.
5048 nics = [n.ToDict() for n in self.nics]
5049 ial = IAllocator(self.cfg, self.rpc,
5050 mode=constants.IALLOCATOR_MODE_ALLOC,
5051 name=self.op.instance_name,
5052 disk_template=self.op.disk_template,
5055 vcpus=self.be_full[constants.BE_VCPUS],
5056 mem_size=self.be_full[constants.BE_MEMORY],
5059 hypervisor=self.op.hypervisor,
5062 ial.Run(self.op.iallocator)
5065 raise errors.OpPrereqError("Can't compute nodes using"
5066 " iallocator '%s': %s" % (self.op.iallocator,
5068 if len(ial.nodes) != ial.required_nodes:
5069 raise errors.OpPrereqError("iallocator '%s' returned invalid number"
5070 " of nodes (%s), required %s" %
5071 (self.op.iallocator, len(ial.nodes),
5072 ial.required_nodes))
5073 self.op.pnode = ial.nodes[0]
5074 self.LogInfo("Selected nodes for instance %s via iallocator %s: %s",
5075 self.op.instance_name, self.op.iallocator,
5076 ", ".join(ial.nodes))
5077 if ial.required_nodes == 2:
5078 self.op.snode = ial.nodes[1]
5080 def BuildHooksEnv(self):
5083 This runs on master, primary and secondary nodes of the instance.
5087 "ADD_MODE": self.op.mode,
5089 if self.op.mode == constants.INSTANCE_IMPORT:
5090 env["SRC_NODE"] = self.op.src_node
5091 env["SRC_PATH"] = self.op.src_path
5092 env["SRC_IMAGES"] = self.src_images
5094 env.update(_BuildInstanceHookEnv(
5095 name=self.op.instance_name,
5096 primary_node=self.op.pnode,
5097 secondary_nodes=self.secondaries,
5098 status=self.op.start,
5099 os_type=self.op.os_type,
5100 memory=self.be_full[constants.BE_MEMORY],
5101 vcpus=self.be_full[constants.BE_VCPUS],
5102 nics=_NICListToTuple(self, self.nics),
5103 disk_template=self.op.disk_template,
5104 disks=[(d["size"], d["mode"]) for d in self.disks],
5107 hypervisor_name=self.op.hypervisor,
5110 nl = ([self.cfg.GetMasterNode(), self.op.pnode] +
5115 def CheckPrereq(self):
5116 """Check prerequisites.
5119 if (not self.cfg.GetVGName() and
5120 self.op.disk_template not in constants.DTS_NOT_LVM):
5121 raise errors.OpPrereqError("Cluster does not support lvm-based"
5124 if self.op.mode == constants.INSTANCE_IMPORT:
5125 src_node = self.op.src_node
5126 src_path = self.op.src_path
5128 if src_node is None:
5129 locked_nodes = self.acquired_locks[locking.LEVEL_NODE]
5130 exp_list = self.rpc.call_export_list(locked_nodes)
5132 for node in exp_list:
5133 if exp_list[node].fail_msg:
5135 if src_path in exp_list[node].payload:
5137 self.op.src_node = src_node = node
5138 self.op.src_path = src_path = os.path.join(constants.EXPORT_DIR,
5142 raise errors.OpPrereqError("No export found for relative path %s" %
5145 _CheckNodeOnline(self, src_node)
5146 result = self.rpc.call_export_info(src_node, src_path)
5147 result.Raise("No export or invalid export found in dir %s" % src_path)
5149 export_info = objects.SerializableConfigParser.Loads(str(result.payload))
5150 if not export_info.has_section(constants.INISECT_EXP):
5151 raise errors.ProgrammerError("Corrupted export config")
5153 ei_version = export_info.get(constants.INISECT_EXP, 'version')
5154 if (int(ei_version) != constants.EXPORT_VERSION):
5155 raise errors.OpPrereqError("Wrong export version %s (wanted %d)" %
5156 (ei_version, constants.EXPORT_VERSION))
5158 # Check that the new instance doesn't have less disks than the export
5159 instance_disks = len(self.disks)
5160 export_disks = export_info.getint(constants.INISECT_INS, 'disk_count')
5161 if instance_disks < export_disks:
5162 raise errors.OpPrereqError("Not enough disks to import."
5163 " (instance: %d, export: %d)" %
5164 (instance_disks, export_disks))
5166 self.op.os_type = export_info.get(constants.INISECT_EXP, 'os')
5168 for idx in range(export_disks):
5169 option = 'disk%d_dump' % idx
5170 if export_info.has_option(constants.INISECT_INS, option):
5171 # FIXME: are the old os-es, disk sizes, etc. useful?
5172 export_name = export_info.get(constants.INISECT_INS, option)
5173 image = os.path.join(src_path, export_name)
5174 disk_images.append(image)
5176 disk_images.append(False)
5178 self.src_images = disk_images
5180 old_name = export_info.get(constants.INISECT_INS, 'name')
5181 # FIXME: int() here could throw a ValueError on broken exports
5182 exp_nic_count = int(export_info.get(constants.INISECT_INS, 'nic_count'))
5183 if self.op.instance_name == old_name:
5184 for idx, nic in enumerate(self.nics):
5185 if nic.mac == constants.VALUE_AUTO and exp_nic_count >= idx:
5186 nic_mac_ini = 'nic%d_mac' % idx
5187 nic.mac = export_info.get(constants.INISECT_INS, nic_mac_ini)
5189 # ENDIF: self.op.mode == constants.INSTANCE_IMPORT
5190 # ip ping checks (we use the same ip that was resolved in ExpandNames)
5191 if self.op.start and not self.op.ip_check:
5192 raise errors.OpPrereqError("Cannot ignore IP address conflicts when"
5193 " adding an instance in start mode")
5195 if self.op.ip_check:
5196 if utils.TcpPing(self.check_ip, constants.DEFAULT_NODED_PORT):
5197 raise errors.OpPrereqError("IP %s of instance %s already in use" %
5198 (self.check_ip, self.op.instance_name))
5200 #### mac address generation
5201 # By generating here the mac address both the allocator and the hooks get
5202 # the real final mac address rather than the 'auto' or 'generate' value.
5203 # There is a race condition between the generation and the instance object
5204 # creation, which means that we know the mac is valid now, but we're not
5205 # sure it will be when we actually add the instance. If things go bad
5206 # adding the instance will abort because of a duplicate mac, and the
5207 # creation job will fail.
5208 for nic in self.nics:
5209 if nic.mac in (constants.VALUE_AUTO, constants.VALUE_GENERATE):
5210 nic.mac = self.cfg.GenerateMAC()
5214 if self.op.iallocator is not None:
5215 self._RunAllocator()
5217 #### node related checks
5219 # check primary node
5220 self.pnode = pnode = self.cfg.GetNodeInfo(self.op.pnode)
5221 assert self.pnode is not None, \
5222 "Cannot retrieve locked node %s" % self.op.pnode
5224 raise errors.OpPrereqError("Cannot use offline primary node '%s'" %
5227 raise errors.OpPrereqError("Cannot use drained primary node '%s'" %
5230 self.secondaries = []
5232 # mirror node verification
5233 if self.op.disk_template in constants.DTS_NET_MIRROR:
5234 if self.op.snode is None:
5235 raise errors.OpPrereqError("The networked disk templates need"
5237 if self.op.snode == pnode.name:
5238 raise errors.OpPrereqError("The secondary node cannot be"
5239 " the primary node.")
5240 _CheckNodeOnline(self, self.op.snode)
5241 _CheckNodeNotDrained(self, self.op.snode)
5242 self.secondaries.append(self.op.snode)
5244 nodenames = [pnode.name] + self.secondaries
5246 req_size = _ComputeDiskSize(self.op.disk_template,
5249 # Check lv size requirements
5250 if req_size is not None:
5251 nodeinfo = self.rpc.call_node_info(nodenames, self.cfg.GetVGName(),
5253 for node in nodenames:
5254 info = nodeinfo[node]
5255 info.Raise("Cannot get current information from node %s" % node)
5257 vg_free = info.get('vg_free', None)
5258 if not isinstance(vg_free, int):
5259 raise errors.OpPrereqError("Can't compute free disk space on"
5261 if req_size > vg_free:
5262 raise errors.OpPrereqError("Not enough disk space on target node %s."
5263 " %d MB available, %d MB required" %
5264 (node, vg_free, req_size))
5266 _CheckHVParams(self, nodenames, self.op.hypervisor, self.op.hvparams)
5269 result = self.rpc.call_os_get(pnode.name, self.op.os_type)
5270 result.Raise("OS '%s' not in supported os list for primary node %s" %
5271 (self.op.os_type, pnode.name), prereq=True)
5273 _CheckNicsBridgesExist(self, self.nics, self.pnode.name)
5275 # memory check on primary node
5277 _CheckNodeFreeMemory(self, self.pnode.name,
5278 "creating instance %s" % self.op.instance_name,
5279 self.be_full[constants.BE_MEMORY],
5282 self.dry_run_result = list(nodenames)
5284 def Exec(self, feedback_fn):
5285 """Create and add the instance to the cluster.
5288 instance = self.op.instance_name
5289 pnode_name = self.pnode.name
5291 ht_kind = self.op.hypervisor
5292 if ht_kind in constants.HTS_REQ_PORT:
5293 network_port = self.cfg.AllocatePort()
5297 ##if self.op.vnc_bind_address is None:
5298 ## self.op.vnc_bind_address = constants.VNC_DEFAULT_BIND_ADDRESS
5300 # this is needed because os.path.join does not accept None arguments
5301 if self.op.file_storage_dir is None:
5302 string_file_storage_dir = ""
5304 string_file_storage_dir = self.op.file_storage_dir
5306 # build the full file storage dir path
5307 file_storage_dir = os.path.normpath(os.path.join(
5308 self.cfg.GetFileStorageDir(),
5309 string_file_storage_dir, instance))
5312 disks = _GenerateDiskTemplate(self,
5313 self.op.disk_template,
5314 instance, pnode_name,
5318 self.op.file_driver,
5321 iobj = objects.Instance(name=instance, os=self.op.os_type,
5322 primary_node=pnode_name,
5323 nics=self.nics, disks=disks,
5324 disk_template=self.op.disk_template,
5326 network_port=network_port,
5327 beparams=self.op.beparams,
5328 hvparams=self.op.hvparams,
5329 hypervisor=self.op.hypervisor,
5332 feedback_fn("* creating instance disks...")
5334 _CreateDisks(self, iobj)
5335 except errors.OpExecError:
5336 self.LogWarning("Device creation failed, reverting...")
5338 _RemoveDisks(self, iobj)
5340 self.cfg.ReleaseDRBDMinors(instance)
5343 feedback_fn("adding instance %s to cluster config" % instance)
5345 self.cfg.AddInstance(iobj)
5346 # Declare that we don't want to remove the instance lock anymore, as we've
5347 # added the instance to the config
5348 del self.remove_locks[locking.LEVEL_INSTANCE]
5349 # Unlock all the nodes
5350 if self.op.mode == constants.INSTANCE_IMPORT:
5351 nodes_keep = [self.op.src_node]
5352 nodes_release = [node for node in self.acquired_locks[locking.LEVEL_NODE]
5353 if node != self.op.src_node]
5354 self.context.glm.release(locking.LEVEL_NODE, nodes_release)
5355 self.acquired_locks[locking.LEVEL_NODE] = nodes_keep
5357 self.context.glm.release(locking.LEVEL_NODE)
5358 del self.acquired_locks[locking.LEVEL_NODE]
5360 if self.op.wait_for_sync:
5361 disk_abort = not _WaitForSync(self, iobj)
5362 elif iobj.disk_template in constants.DTS_NET_MIRROR:
5363 # make sure the disks are not degraded (still sync-ing is ok)
5365 feedback_fn("* checking mirrors status")
5366 disk_abort = not _WaitForSync(self, iobj, oneshot=True)
5371 _RemoveDisks(self, iobj)
5372 self.cfg.RemoveInstance(iobj.name)
5373 # Make sure the instance lock gets removed
5374 self.remove_locks[locking.LEVEL_INSTANCE] = iobj.name
5375 raise errors.OpExecError("There are some degraded disks for"
5378 feedback_fn("creating os for instance %s on node %s" %
5379 (instance, pnode_name))
5381 if iobj.disk_template != constants.DT_DISKLESS:
5382 if self.op.mode == constants.INSTANCE_CREATE:
5383 feedback_fn("* running the instance OS create scripts...")
5384 result = self.rpc.call_instance_os_add(pnode_name, iobj, False)
5385 result.Raise("Could not add os for instance %s"
5386 " on node %s" % (instance, pnode_name))
5388 elif self.op.mode == constants.INSTANCE_IMPORT:
5389 feedback_fn("* running the instance OS import scripts...")
5390 src_node = self.op.src_node
5391 src_images = self.src_images
5392 cluster_name = self.cfg.GetClusterName()
5393 import_result = self.rpc.call_instance_os_import(pnode_name, iobj,
5394 src_node, src_images,
5396 msg = import_result.fail_msg
5398 self.LogWarning("Error while importing the disk images for instance"
5399 " %s on node %s: %s" % (instance, pnode_name, msg))
5401 # also checked in the prereq part
5402 raise errors.ProgrammerError("Unknown OS initialization mode '%s'"
5406 iobj.admin_up = True
5407 self.cfg.Update(iobj)
5408 logging.info("Starting instance %s on node %s", instance, pnode_name)
5409 feedback_fn("* starting instance...")
5410 result = self.rpc.call_instance_start(pnode_name, iobj, None, None)
5411 result.Raise("Could not start instance")
5413 return list(iobj.all_nodes)
5416 class LUConnectConsole(NoHooksLU):
5417 """Connect to an instance's console.
5419 This is somewhat special in that it returns the command line that
5420 you need to run on the master node in order to connect to the
5424 _OP_REQP = ["instance_name"]
5427 def ExpandNames(self):
5428 self._ExpandAndLockInstance()
5430 def CheckPrereq(self):
5431 """Check prerequisites.
5433 This checks that the instance is in the cluster.
5436 self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
5437 assert self.instance is not None, \
5438 "Cannot retrieve locked instance %s" % self.op.instance_name
5439 _CheckNodeOnline(self, self.instance.primary_node)
5441 def Exec(self, feedback_fn):
5442 """Connect to the console of an instance
5445 instance = self.instance
5446 node = instance.primary_node
5448 node_insts = self.rpc.call_instance_list([node],
5449 [instance.hypervisor])[node]
5450 node_insts.Raise("Can't get node information from %s" % node)
5452 if instance.name not in node_insts.payload:
5453 raise errors.OpExecError("Instance %s is not running." % instance.name)
5455 logging.debug("Connecting to console of %s on %s", instance.name, node)
5457 hyper = hypervisor.GetHypervisor(instance.hypervisor)
5458 cluster = self.cfg.GetClusterInfo()
5459 # beparams and hvparams are passed separately, to avoid editing the
5460 # instance and then saving the defaults in the instance itself.
5461 hvparams = cluster.FillHV(instance)
5462 beparams = cluster.FillBE(instance)
5463 console_cmd = hyper.GetShellCommandForConsole(instance, hvparams, beparams)
5466 return self.ssh.BuildCmd(node, "root", console_cmd, batch=True, tty=True)
5469 class LUReplaceDisks(LogicalUnit):
5470 """Replace the disks of an instance.
5473 HPATH = "mirrors-replace"
5474 HTYPE = constants.HTYPE_INSTANCE
5475 _OP_REQP = ["instance_name", "mode", "disks"]
5478 def CheckArguments(self):
5479 if not hasattr(self.op, "remote_node"):
5480 self.op.remote_node = None
5481 if not hasattr(self.op, "iallocator"):
5482 self.op.iallocator = None
5484 TLReplaceDisks.CheckArguments(self.op.mode, self.op.remote_node,
5487 def ExpandNames(self):
5488 self._ExpandAndLockInstance()
5490 if self.op.iallocator is not None:
5491 self.needed_locks[locking.LEVEL_NODE] = locking.ALL_SET
5493 elif self.op.remote_node is not None:
5494 remote_node = self.cfg.ExpandNodeName(self.op.remote_node)
5495 if remote_node is None:
5496 raise errors.OpPrereqError("Node '%s' not known" %
5497 self.op.remote_node)
5499 self.op.remote_node = remote_node
5501 # Warning: do not remove the locking of the new secondary here
5502 # unless DRBD8.AddChildren is changed to work in parallel;
5503 # currently it doesn't since parallel invocations of
5504 # FindUnusedMinor will conflict
5505 self.needed_locks[locking.LEVEL_NODE] = [remote_node]
5506 self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_APPEND
5509 self.needed_locks[locking.LEVEL_NODE] = []
5510 self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
5512 self.replacer = TLReplaceDisks(self, self.op.instance_name, self.op.mode,
5513 self.op.iallocator, self.op.remote_node,
5516 self.tasklets = [self.replacer]
5518 def DeclareLocks(self, level):
5519 # If we're not already locking all nodes in the set we have to declare the
5520 # instance's primary/secondary nodes.
5521 if (level == locking.LEVEL_NODE and
5522 self.needed_locks[locking.LEVEL_NODE] is not locking.ALL_SET):
5523 self._LockInstancesNodes()
5525 def BuildHooksEnv(self):
5528 This runs on the master, the primary and all the secondaries.
5531 instance = self.replacer.instance
5533 "MODE": self.op.mode,
5534 "NEW_SECONDARY": self.op.remote_node,
5535 "OLD_SECONDARY": instance.secondary_nodes[0],
5537 env.update(_BuildInstanceHookEnvByObject(self, instance))
5539 self.cfg.GetMasterNode(),
5540 instance.primary_node,
5542 if self.op.remote_node is not None:
5543 nl.append(self.op.remote_node)
5547 class LUEvacuateNode(LogicalUnit):
5548 """Relocate the secondary instances from a node.
5551 HPATH = "node-evacuate"
5552 HTYPE = constants.HTYPE_NODE
5553 _OP_REQP = ["node_name"]
5556 def CheckArguments(self):
5557 if not hasattr(self.op, "remote_node"):
5558 self.op.remote_node = None
5559 if not hasattr(self.op, "iallocator"):
5560 self.op.iallocator = None
5562 TLReplaceDisks.CheckArguments(constants.REPLACE_DISK_CHG,
5563 self.op.remote_node,
5566 def ExpandNames(self):
5567 self.op.node_name = self.cfg.ExpandNodeName(self.op.node_name)
5568 if self.op.node_name is None:
5569 raise errors.OpPrereqError("Node '%s' not known" % self.op.node_name)
5571 self.needed_locks = {}
5573 # Declare node locks
5574 if self.op.iallocator is not None:
5575 self.needed_locks[locking.LEVEL_NODE] = locking.ALL_SET
5577 elif self.op.remote_node is not None:
5578 remote_node = self.cfg.ExpandNodeName(self.op.remote_node)
5579 if remote_node is None:
5580 raise errors.OpPrereqError("Node '%s' not known" %
5581 self.op.remote_node)
5583 self.op.remote_node = remote_node
5585 # Warning: do not remove the locking of the new secondary here
5586 # unless DRBD8.AddChildren is changed to work in parallel;
5587 # currently it doesn't since parallel invocations of
5588 # FindUnusedMinor will conflict
5589 self.needed_locks[locking.LEVEL_NODE] = [remote_node]
5590 self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_APPEND
5593 raise errors.OpPrereqError("Invalid parameters")
5595 # Create tasklets for replacing disks for all secondary instances on this
5600 for inst in _GetNodeSecondaryInstances(self.cfg, self.op.node_name):
5601 logging.debug("Replacing disks for instance %s", inst.name)
5602 names.append(inst.name)
5604 replacer = TLReplaceDisks(self, inst.name, constants.REPLACE_DISK_CHG,
5605 self.op.iallocator, self.op.remote_node, [])
5606 tasklets.append(replacer)
5608 self.tasklets = tasklets
5609 self.instance_names = names
5611 # Declare instance locks
5612 self.needed_locks[locking.LEVEL_INSTANCE] = self.instance_names
5614 def DeclareLocks(self, level):
5615 # If we're not already locking all nodes in the set we have to declare the
5616 # instance's primary/secondary nodes.
5617 if (level == locking.LEVEL_NODE and
5618 self.needed_locks[locking.LEVEL_NODE] is not locking.ALL_SET):
5619 self._LockInstancesNodes()
5621 def BuildHooksEnv(self):
5624 This runs on the master, the primary and all the secondaries.
5628 "NODE_NAME": self.op.node_name,
5631 nl = [self.cfg.GetMasterNode()]
5633 if self.op.remote_node is not None:
5634 env["NEW_SECONDARY"] = self.op.remote_node
5635 nl.append(self.op.remote_node)
5637 return (env, nl, nl)
5640 class TLReplaceDisks(Tasklet):
5641 """Replaces disks for an instance.
5643 Note: Locking is not within the scope of this class.
5646 def __init__(self, lu, instance_name, mode, iallocator_name, remote_node,
5648 """Initializes this class.
5651 Tasklet.__init__(self, lu)
5654 self.instance_name = instance_name
5656 self.iallocator_name = iallocator_name
5657 self.remote_node = remote_node
5661 self.instance = None
5662 self.new_node = None
5663 self.target_node = None
5664 self.other_node = None
5665 self.remote_node_info = None
5666 self.node_secondary_ip = None
5669 def CheckArguments(mode, remote_node, iallocator):
5670 """Helper function for users of this class.
5673 # check for valid parameter combination
5674 if mode == constants.REPLACE_DISK_CHG:
5675 if remote_node is None and iallocator is None:
5676 raise errors.OpPrereqError("When changing the secondary either an"
5677 " iallocator script must be used or the"
5680 if remote_node is not None and iallocator is not None:
5681 raise errors.OpPrereqError("Give either the iallocator or the new"
5682 " secondary, not both")
5684 elif remote_node is not None or iallocator is not None:
5685 # Not replacing the secondary
5686 raise errors.OpPrereqError("The iallocator and new node options can"
5687 " only be used when changing the"
5691 def _RunAllocator(lu, iallocator_name, instance_name, relocate_from):
5692 """Compute a new secondary node using an IAllocator.
5695 ial = IAllocator(lu.cfg, lu.rpc,
5696 mode=constants.IALLOCATOR_MODE_RELOC,
5698 relocate_from=relocate_from)
5700 ial.Run(iallocator_name)
5703 raise errors.OpPrereqError("Can't compute nodes using iallocator '%s':"
5704 " %s" % (iallocator_name, ial.info))
5706 if len(ial.nodes) != ial.required_nodes:
5707 raise errors.OpPrereqError("iallocator '%s' returned invalid number"
5708 " of nodes (%s), required %s" %
5709 (len(ial.nodes), ial.required_nodes))
5711 remote_node_name = ial.nodes[0]
5713 lu.LogInfo("Selected new secondary for instance '%s': %s",
5714 instance_name, remote_node_name)
5716 return remote_node_name
5718 def _FindFaultyDisks(self, node_name):
5721 for dev in self.instance.disks:
5722 self.cfg.SetDiskID(dev, node_name)
5724 result = self.rpc.call_blockdev_getmirrorstatus(node_name,
5725 self.instance.disks)
5726 result.Raise("Failed to get disk status from node %s" % node_name,
5729 for idx, bdev_status in enumerate(result.payload):
5730 if bdev_status and bdev_status.ldisk_status == constants.LDS_FAULTY:
5735 def CheckPrereq(self):
5736 """Check prerequisites.
5738 This checks that the instance is in the cluster.
5741 self.instance = self.cfg.GetInstanceInfo(self.instance_name)
5742 assert self.instance is not None, \
5743 "Cannot retrieve locked instance %s" % self.instance_name
5745 if self.instance.disk_template != constants.DT_DRBD8:
5746 raise errors.OpPrereqError("Can only run replace disks for DRBD8-based"
5749 if len(self.instance.secondary_nodes) != 1:
5750 raise errors.OpPrereqError("The instance has a strange layout,"
5751 " expected one secondary but found %d" %
5752 len(self.instance.secondary_nodes))
5754 secondary_node = self.instance.secondary_nodes[0]
5756 if self.iallocator_name is None:
5757 remote_node = self.remote_node
5759 remote_node = self._RunAllocator(self.lu, self.iallocator_name,
5760 self.instance.name, secondary_node)
5762 if remote_node is not None:
5763 self.remote_node_info = self.cfg.GetNodeInfo(remote_node)
5764 assert self.remote_node_info is not None, \
5765 "Cannot retrieve locked node %s" % remote_node
5767 self.remote_node_info = None
5769 if remote_node == self.instance.primary_node:
5770 raise errors.OpPrereqError("The specified node is the primary node of"
5773 if remote_node == secondary_node:
5774 raise errors.OpPrereqError("The specified node is already the"
5775 " secondary node of the instance.")
5777 if self.mode == constants.REPLACE_DISK_AUTO:
5779 raise errors.OpPrereqError("Cannot specify disks to be replaced")
5781 faulty_primary = self._FindFaultyDisks(self.instance.primary_node)
5782 faulty_secondary = self._FindFaultyDisks(secondary_node)
5784 if faulty_primary and faulty_secondary:
5785 raise errors.OpPrereqError("Instance %s has faulty disks on more than"
5786 " one node and can not be repaired"
5787 " automatically" % self.instance_name)
5790 self.disks = faulty_primary
5791 self.target_node = self.instance.primary_node
5792 self.other_node = secondary_node
5793 check_nodes = [self.target_node, self.other_node]
5794 elif faulty_secondary:
5795 self.disks = faulty_secondary
5796 self.target_node = secondary_node
5797 self.other_node = self.instance.primary_node
5798 check_nodes = [self.target_node, self.other_node]
5804 # Non-automatic modes
5805 if self.mode == constants.REPLACE_DISK_PRI:
5806 self.target_node = self.instance.primary_node
5807 self.other_node = secondary_node
5808 check_nodes = [self.target_node, self.other_node]
5810 elif self.mode == constants.REPLACE_DISK_SEC:
5811 self.target_node = secondary_node
5812 self.other_node = self.instance.primary_node
5813 check_nodes = [self.target_node, self.other_node]
5815 elif self.mode == constants.REPLACE_DISK_CHG:
5816 self.new_node = remote_node
5817 self.other_node = self.instance.primary_node
5818 self.target_node = secondary_node
5819 check_nodes = [self.new_node, self.other_node]
5821 _CheckNodeNotDrained(self.lu, remote_node)
5824 raise errors.ProgrammerError("Unhandled disk replace mode (%s)" %
5827 # If not specified all disks should be replaced
5829 self.disks = range(len(self.instance.disks))
5831 for node in check_nodes:
5832 _CheckNodeOnline(self.lu, node)
5834 # Check whether disks are valid
5835 for disk_idx in self.disks:
5836 self.instance.FindDisk(disk_idx)
5838 # Get secondary node IP addresses
5841 for node_name in [self.target_node, self.other_node, self.new_node]:
5842 if node_name is not None:
5843 node_2nd_ip[node_name] = self.cfg.GetNodeInfo(node_name).secondary_ip
5845 self.node_secondary_ip = node_2nd_ip
5847 def Exec(self, feedback_fn):
5848 """Execute disk replacement.
5850 This dispatches the disk replacement to the appropriate handler.
5854 feedback_fn("No disks need replacement")
5857 feedback_fn("Replacing disk(s) %s for %s" %
5858 (", ".join([str(i) for i in self.disks]), self.instance.name))
5860 activate_disks = (not self.instance.admin_up)
5862 # Activate the instance disks if we're replacing them on a down instance
5864 _StartInstanceDisks(self.lu, self.instance, True)
5867 # Should we replace the secondary node?
5868 if self.new_node is not None:
5869 return self._ExecDrbd8Secondary()
5871 return self._ExecDrbd8DiskOnly()
5874 # Deactivate the instance disks if we're replacing them on a down instance
5876 _SafeShutdownInstanceDisks(self.lu, self.instance)
5878 def _CheckVolumeGroup(self, nodes):
5879 self.lu.LogInfo("Checking volume groups")
5881 vgname = self.cfg.GetVGName()
5883 # Make sure volume group exists on all involved nodes
5884 results = self.rpc.call_vg_list(nodes)
5886 raise errors.OpExecError("Can't list volume groups on the nodes")
5890 res.Raise("Error checking node %s" % node)
5891 if vgname not in res.payload:
5892 raise errors.OpExecError("Volume group '%s' not found on node %s" %
5895 def _CheckDisksExistence(self, nodes):
5896 # Check disk existence
5897 for idx, dev in enumerate(self.instance.disks):
5898 if idx not in self.disks:
5902 self.lu.LogInfo("Checking disk/%d on %s" % (idx, node))
5903 self.cfg.SetDiskID(dev, node)
5905 result = self.rpc.call_blockdev_find(node, dev)
5907 msg = result.fail_msg
5908 if msg or not result.payload:
5910 msg = "disk not found"
5911 raise errors.OpExecError("Can't find disk/%d on node %s: %s" %
5914 def _CheckDisksConsistency(self, node_name, on_primary, ldisk):
5915 for idx, dev in enumerate(self.instance.disks):
5916 if idx not in self.disks:
5919 self.lu.LogInfo("Checking disk/%d consistency on node %s" %
5922 if not _CheckDiskConsistency(self.lu, dev, node_name, on_primary,
5924 raise errors.OpExecError("Node %s has degraded storage, unsafe to"
5925 " replace disks for instance %s" %
5926 (node_name, self.instance.name))
5928 def _CreateNewStorage(self, node_name):
5929 vgname = self.cfg.GetVGName()
5932 for idx, dev in enumerate(self.instance.disks):
5933 if idx not in self.disks:
5936 self.lu.LogInfo("Adding storage on %s for disk/%d" % (node_name, idx))
5938 self.cfg.SetDiskID(dev, node_name)
5940 lv_names = [".disk%d_%s" % (idx, suffix) for suffix in ["data", "meta"]]
5941 names = _GenerateUniqueNames(self.lu, lv_names)
5943 lv_data = objects.Disk(dev_type=constants.LD_LV, size=dev.size,
5944 logical_id=(vgname, names[0]))
5945 lv_meta = objects.Disk(dev_type=constants.LD_LV, size=128,
5946 logical_id=(vgname, names[1]))
5948 new_lvs = [lv_data, lv_meta]
5949 old_lvs = dev.children
5950 iv_names[dev.iv_name] = (dev, old_lvs, new_lvs)
5952 # we pass force_create=True to force the LVM creation
5953 for new_lv in new_lvs:
5954 _CreateBlockDev(self.lu, node_name, self.instance, new_lv, True,
5955 _GetInstanceInfoText(self.instance), False)
5959 def _CheckDevices(self, node_name, iv_names):
5960 for name, (dev, old_lvs, new_lvs) in iv_names.iteritems():
5961 self.cfg.SetDiskID(dev, node_name)
5963 result = self.rpc.call_blockdev_find(node_name, dev)
5965 msg = result.fail_msg
5966 if msg or not result.payload:
5968 msg = "disk not found"
5969 raise errors.OpExecError("Can't find DRBD device %s: %s" %
5972 if result.payload.is_degraded:
5973 raise errors.OpExecError("DRBD device %s is degraded!" % name)
5975 def _RemoveOldStorage(self, node_name, iv_names):
5976 for name, (dev, old_lvs, _) in iv_names.iteritems():
5977 self.lu.LogInfo("Remove logical volumes for %s" % name)
5980 self.cfg.SetDiskID(lv, node_name)
5982 msg = self.rpc.call_blockdev_remove(node_name, lv).fail_msg
5984 self.lu.LogWarning("Can't remove old LV: %s" % msg,
5985 hint="remove unused LVs manually")
5987 def _ExecDrbd8DiskOnly(self):
5988 """Replace a disk on the primary or secondary for DRBD 8.
5990 The algorithm for replace is quite complicated:
5992 1. for each disk to be replaced:
5994 1. create new LVs on the target node with unique names
5995 1. detach old LVs from the drbd device
5996 1. rename old LVs to name_replaced.<time_t>
5997 1. rename new LVs to old LVs
5998 1. attach the new LVs (with the old names now) to the drbd device
6000 1. wait for sync across all devices
6002 1. for each modified disk:
6004 1. remove old LVs (which have the name name_replaces.<time_t>)
6006 Failures are not very well handled.
6011 # Step: check device activation
6012 self.lu.LogStep(1, steps_total, "Check device existence")
6013 self._CheckDisksExistence([self.other_node, self.target_node])
6014 self._CheckVolumeGroup([self.target_node, self.other_node])
6016 # Step: check other node consistency
6017 self.lu.LogStep(2, steps_total, "Check peer consistency")
6018 self._CheckDisksConsistency(self.other_node,
6019 self.other_node == self.instance.primary_node,
6022 # Step: create new storage
6023 self.lu.LogStep(3, steps_total, "Allocate new storage")
6024 iv_names = self._CreateNewStorage(self.target_node)
6026 # Step: for each lv, detach+rename*2+attach
6027 self.lu.LogStep(4, steps_total, "Changing drbd configuration")
6028 for dev, old_lvs, new_lvs in iv_names.itervalues():
6029 self.lu.LogInfo("Detaching %s drbd from local storage" % dev.iv_name)
6031 result = self.rpc.call_blockdev_removechildren(self.target_node, dev, old_lvs)
6032 result.Raise("Can't detach drbd from local storage on node"
6033 " %s for device %s" % (self.target_node, dev.iv_name))
6035 #cfg.Update(instance)
6037 # ok, we created the new LVs, so now we know we have the needed
6038 # storage; as such, we proceed on the target node to rename
6039 # old_lv to _old, and new_lv to old_lv; note that we rename LVs
6040 # using the assumption that logical_id == physical_id (which in
6041 # turn is the unique_id on that node)
6043 # FIXME(iustin): use a better name for the replaced LVs
6044 temp_suffix = int(time.time())
6045 ren_fn = lambda d, suff: (d.physical_id[0],
6046 d.physical_id[1] + "_replaced-%s" % suff)
6048 # Build the rename list based on what LVs exist on the node
6049 rename_old_to_new = []
6050 for to_ren in old_lvs:
6051 result = self.rpc.call_blockdev_find(self.target_node, to_ren)
6052 if not result.fail_msg and result.payload:
6054 rename_old_to_new.append((to_ren, ren_fn(to_ren, temp_suffix)))
6056 self.lu.LogInfo("Renaming the old LVs on the target node")
6057 result = self.rpc.call_blockdev_rename(self.target_node, rename_old_to_new)
6058 result.Raise("Can't rename old LVs on node %s" % self.target_node)
6060 # Now we rename the new LVs to the old LVs
6061 self.lu.LogInfo("Renaming the new LVs on the target node")
6062 rename_new_to_old = [(new, old.physical_id)
6063 for old, new in zip(old_lvs, new_lvs)]
6064 result = self.rpc.call_blockdev_rename(self.target_node, rename_new_to_old)
6065 result.Raise("Can't rename new LVs on node %s" % self.target_node)
6067 for old, new in zip(old_lvs, new_lvs):
6068 new.logical_id = old.logical_id
6069 self.cfg.SetDiskID(new, self.target_node)
6071 for disk in old_lvs:
6072 disk.logical_id = ren_fn(disk, temp_suffix)
6073 self.cfg.SetDiskID(disk, self.target_node)
6075 # Now that the new lvs have the old name, we can add them to the device
6076 self.lu.LogInfo("Adding new mirror component on %s" % self.target_node)
6077 result = self.rpc.call_blockdev_addchildren(self.target_node, dev, new_lvs)
6078 msg = result.fail_msg
6080 for new_lv in new_lvs:
6081 msg2 = self.rpc.call_blockdev_remove(self.target_node, new_lv).fail_msg
6083 self.lu.LogWarning("Can't rollback device %s: %s", dev, msg2,
6084 hint=("cleanup manually the unused logical"
6086 raise errors.OpExecError("Can't add local storage to drbd: %s" % msg)
6088 dev.children = new_lvs
6090 self.cfg.Update(self.instance)
6093 # This can fail as the old devices are degraded and _WaitForSync
6094 # does a combined result over all disks, so we don't check its return value
6095 self.lu.LogStep(5, steps_total, "Sync devices")
6096 _WaitForSync(self.lu, self.instance, unlock=True)
6098 # Check all devices manually
6099 self._CheckDevices(self.instance.primary_node, iv_names)
6101 # Step: remove old storage
6102 self.lu.LogStep(6, steps_total, "Removing old storage")
6103 self._RemoveOldStorage(self.target_node, iv_names)
6105 def _ExecDrbd8Secondary(self):
6106 """Replace the secondary node for DRBD 8.
6108 The algorithm for replace is quite complicated:
6109 - for all disks of the instance:
6110 - create new LVs on the new node with same names
6111 - shutdown the drbd device on the old secondary
6112 - disconnect the drbd network on the primary
6113 - create the drbd device on the new secondary
6114 - network attach the drbd on the primary, using an artifice:
6115 the drbd code for Attach() will connect to the network if it
6116 finds a device which is connected to the good local disks but
6118 - wait for sync across all devices
6119 - remove all disks from the old secondary
6121 Failures are not very well handled.
6126 # Step: check device activation
6127 self.lu.LogStep(1, steps_total, "Check device existence")
6128 self._CheckDisksExistence([self.instance.primary_node])
6129 self._CheckVolumeGroup([self.instance.primary_node])
6131 # Step: check other node consistency
6132 self.lu.LogStep(2, steps_total, "Check peer consistency")
6133 self._CheckDisksConsistency(self.instance.primary_node, True, True)
6135 # Step: create new storage
6136 self.lu.LogStep(3, steps_total, "Allocate new storage")
6137 for idx, dev in enumerate(self.instance.disks):
6138 self.lu.LogInfo("Adding new local storage on %s for disk/%d" %
6139 (self.new_node, idx))
6140 # we pass force_create=True to force LVM creation
6141 for new_lv in dev.children:
6142 _CreateBlockDev(self.lu, self.new_node, self.instance, new_lv, True,
6143 _GetInstanceInfoText(self.instance), False)
6145 # Step 4: dbrd minors and drbd setups changes
6146 # after this, we must manually remove the drbd minors on both the
6147 # error and the success paths
6148 self.lu.LogStep(4, steps_total, "Changing drbd configuration")
6149 minors = self.cfg.AllocateDRBDMinor([self.new_node for dev in self.instance.disks],
6151 logging.debug("Allocated minors %r" % (minors,))
6154 for idx, (dev, new_minor) in enumerate(zip(self.instance.disks, minors)):
6155 self.lu.LogInfo("activating a new drbd on %s for disk/%d" % (self.new_node, idx))
6156 # create new devices on new_node; note that we create two IDs:
6157 # one without port, so the drbd will be activated without
6158 # networking information on the new node at this stage, and one
6159 # with network, for the latter activation in step 4
6160 (o_node1, o_node2, o_port, o_minor1, o_minor2, o_secret) = dev.logical_id
6161 if self.instance.primary_node == o_node1:
6166 new_alone_id = (self.instance.primary_node, self.new_node, None, p_minor, new_minor, o_secret)
6167 new_net_id = (self.instance.primary_node, self.new_node, o_port, p_minor, new_minor, o_secret)
6169 iv_names[idx] = (dev, dev.children, new_net_id)
6170 logging.debug("Allocated new_minor: %s, new_logical_id: %s", new_minor,
6172 new_drbd = objects.Disk(dev_type=constants.LD_DRBD8,
6173 logical_id=new_alone_id,
6174 children=dev.children,
6177 _CreateSingleBlockDev(self.lu, self.new_node, self.instance, new_drbd,
6178 _GetInstanceInfoText(self.instance), False)
6179 except errors.GenericError:
6180 self.cfg.ReleaseDRBDMinors(self.instance.name)
6183 # We have new devices, shutdown the drbd on the old secondary
6184 for idx, dev in enumerate(self.instance.disks):
6185 self.lu.LogInfo("Shutting down drbd for disk/%d on old node" % idx)
6186 self.cfg.SetDiskID(dev, self.target_node)
6187 msg = self.rpc.call_blockdev_shutdown(self.target_node, dev).fail_msg
6189 self.lu.LogWarning("Failed to shutdown drbd for disk/%d on old"
6190 "node: %s" % (idx, msg),
6191 hint=("Please cleanup this device manually as"
6192 " soon as possible"))
6194 self.lu.LogInfo("Detaching primary drbds from the network (=> standalone)")
6195 result = self.rpc.call_drbd_disconnect_net([self.instance.primary_node], self.node_secondary_ip,
6196 self.instance.disks)[self.instance.primary_node]
6198 msg = result.fail_msg
6200 # detaches didn't succeed (unlikely)
6201 self.cfg.ReleaseDRBDMinors(self.instance.name)
6202 raise errors.OpExecError("Can't detach the disks from the network on"
6203 " old node: %s" % (msg,))
6205 # if we managed to detach at least one, we update all the disks of
6206 # the instance to point to the new secondary
6207 self.lu.LogInfo("Updating instance configuration")
6208 for dev, _, new_logical_id in iv_names.itervalues():
6209 dev.logical_id = new_logical_id
6210 self.cfg.SetDiskID(dev, self.instance.primary_node)
6212 self.cfg.Update(self.instance)
6214 # and now perform the drbd attach
6215 self.lu.LogInfo("Attaching primary drbds to new secondary"
6216 " (standalone => connected)")
6217 result = self.rpc.call_drbd_attach_net([self.instance.primary_node, self.new_node], self.node_secondary_ip,
6218 self.instance.disks, self.instance.name,
6220 for to_node, to_result in result.items():
6221 msg = to_result.fail_msg
6223 self.lu.LogWarning("Can't attach drbd disks on node %s: %s", to_node, msg,
6224 hint=("please do a gnt-instance info to see the"
6225 " status of disks"))
6228 # This can fail as the old devices are degraded and _WaitForSync
6229 # does a combined result over all disks, so we don't check its return value
6230 self.lu.LogStep(5, steps_total, "Sync devices")
6231 _WaitForSync(self.lu, self.instance, unlock=True)
6233 # Check all devices manually
6234 self._CheckDevices(self.instance.primary_node, iv_names)
6236 # Step: remove old storage
6237 self.lu.LogStep(6, steps_total, "Removing old storage")
6238 self._RemoveOldStorage(self.target_node, iv_names)
6241 class LUGrowDisk(LogicalUnit):
6242 """Grow a disk of an instance.
6246 HTYPE = constants.HTYPE_INSTANCE
6247 _OP_REQP = ["instance_name", "disk", "amount", "wait_for_sync"]
6250 def ExpandNames(self):
6251 self._ExpandAndLockInstance()
6252 self.needed_locks[locking.LEVEL_NODE] = []
6253 self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
6255 def DeclareLocks(self, level):
6256 if level == locking.LEVEL_NODE:
6257 self._LockInstancesNodes()
6259 def BuildHooksEnv(self):
6262 This runs on the master, the primary and all the secondaries.
6266 "DISK": self.op.disk,
6267 "AMOUNT": self.op.amount,
6269 env.update(_BuildInstanceHookEnvByObject(self, self.instance))
6271 self.cfg.GetMasterNode(),
6272 self.instance.primary_node,
6276 def CheckPrereq(self):
6277 """Check prerequisites.
6279 This checks that the instance is in the cluster.
6282 instance = self.cfg.GetInstanceInfo(self.op.instance_name)
6283 assert instance is not None, \
6284 "Cannot retrieve locked instance %s" % self.op.instance_name
6285 nodenames = list(instance.all_nodes)
6286 for node in nodenames:
6287 _CheckNodeOnline(self, node)
6290 self.instance = instance
6292 if instance.disk_template not in (constants.DT_PLAIN, constants.DT_DRBD8):
6293 raise errors.OpPrereqError("Instance's disk layout does not support"
6296 self.disk = instance.FindDisk(self.op.disk)
6298 nodeinfo = self.rpc.call_node_info(nodenames, self.cfg.GetVGName(),
6299 instance.hypervisor)
6300 for node in nodenames:
6301 info = nodeinfo[node]
6302 info.Raise("Cannot get current information from node %s" % node)
6303 vg_free = info.payload.get('vg_free', None)
6304 if not isinstance(vg_free, int):
6305 raise errors.OpPrereqError("Can't compute free disk space on"
6307 if self.op.amount > vg_free:
6308 raise errors.OpPrereqError("Not enough disk space on target node %s:"
6309 " %d MiB available, %d MiB required" %
6310 (node, vg_free, self.op.amount))
6312 def Exec(self, feedback_fn):
6313 """Execute disk grow.
6316 instance = self.instance
6318 for node in instance.all_nodes:
6319 self.cfg.SetDiskID(disk, node)
6320 result = self.rpc.call_blockdev_grow(node, disk, self.op.amount)
6321 result.Raise("Grow request failed to node %s" % node)
6322 disk.RecordGrow(self.op.amount)
6323 self.cfg.Update(instance)
6324 if self.op.wait_for_sync:
6325 disk_abort = not _WaitForSync(self, instance)
6327 self.proc.LogWarning("Warning: disk sync-ing has not returned a good"
6328 " status.\nPlease check the instance.")
6331 class LUQueryInstanceData(NoHooksLU):
6332 """Query runtime instance data.
6335 _OP_REQP = ["instances", "static"]
6338 def ExpandNames(self):
6339 self.needed_locks = {}
6340 self.share_locks = dict.fromkeys(locking.LEVELS, 1)
6342 if not isinstance(self.op.instances, list):
6343 raise errors.OpPrereqError("Invalid argument type 'instances'")
6345 if self.op.instances:
6346 self.wanted_names = []
6347 for name in self.op.instances:
6348 full_name = self.cfg.ExpandInstanceName(name)
6349 if full_name is None:
6350 raise errors.OpPrereqError("Instance '%s' not known" % name)
6351 self.wanted_names.append(full_name)
6352 self.needed_locks[locking.LEVEL_INSTANCE] = self.wanted_names
6354 self.wanted_names = None
6355 self.needed_locks[locking.LEVEL_INSTANCE] = locking.ALL_SET
6357 self.needed_locks[locking.LEVEL_NODE] = []
6358 self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
6360 def DeclareLocks(self, level):
6361 if level == locking.LEVEL_NODE:
6362 self._LockInstancesNodes()
6364 def CheckPrereq(self):
6365 """Check prerequisites.
6367 This only checks the optional instance list against the existing names.
6370 if self.wanted_names is None:
6371 self.wanted_names = self.acquired_locks[locking.LEVEL_INSTANCE]
6373 self.wanted_instances = [self.cfg.GetInstanceInfo(name) for name
6374 in self.wanted_names]
6377 def _ComputeBlockdevStatus(self, node, instance_name, dev):
6378 """Returns the status of a block device
6384 self.cfg.SetDiskID(dev, node)
6386 result = self.rpc.call_blockdev_find(node, dev)
6390 result.Raise("Can't compute disk status for %s" % instance_name)
6392 status = result.payload
6396 return (status.dev_path, status.major, status.minor,
6397 status.sync_percent, status.estimated_time,
6398 status.is_degraded, status.ldisk_status)
6400 def _ComputeDiskStatus(self, instance, snode, dev):
6401 """Compute block device status.
6404 if dev.dev_type in constants.LDS_DRBD:
6405 # we change the snode then (otherwise we use the one passed in)
6406 if dev.logical_id[0] == instance.primary_node:
6407 snode = dev.logical_id[1]
6409 snode = dev.logical_id[0]
6411 dev_pstatus = self._ComputeBlockdevStatus(instance.primary_node,
6413 dev_sstatus = self._ComputeBlockdevStatus(snode, instance.name, dev)
6416 dev_children = [self._ComputeDiskStatus(instance, snode, child)
6417 for child in dev.children]
6422 "iv_name": dev.iv_name,
6423 "dev_type": dev.dev_type,
6424 "logical_id": dev.logical_id,
6425 "physical_id": dev.physical_id,
6426 "pstatus": dev_pstatus,
6427 "sstatus": dev_sstatus,
6428 "children": dev_children,
6435 def Exec(self, feedback_fn):
6436 """Gather and return data"""
6439 cluster = self.cfg.GetClusterInfo()
6441 for instance in self.wanted_instances:
6442 if not self.op.static:
6443 remote_info = self.rpc.call_instance_info(instance.primary_node,
6445 instance.hypervisor)
6446 remote_info.Raise("Error checking node %s" % instance.primary_node)
6447 remote_info = remote_info.payload
6448 if remote_info and "state" in remote_info:
6451 remote_state = "down"
6454 if instance.admin_up:
6457 config_state = "down"
6459 disks = [self._ComputeDiskStatus(instance, None, device)
6460 for device in instance.disks]
6463 "name": instance.name,
6464 "config_state": config_state,
6465 "run_state": remote_state,
6466 "pnode": instance.primary_node,
6467 "snodes": instance.secondary_nodes,
6469 # this happens to be the same format used for hooks
6470 "nics": _NICListToTuple(self, instance.nics),
6472 "hypervisor": instance.hypervisor,
6473 "network_port": instance.network_port,
6474 "hv_instance": instance.hvparams,
6475 "hv_actual": cluster.FillHV(instance),
6476 "be_instance": instance.beparams,
6477 "be_actual": cluster.FillBE(instance),
6480 result[instance.name] = idict
6485 class LUSetInstanceParams(LogicalUnit):
6486 """Modifies an instances's parameters.
6489 HPATH = "instance-modify"
6490 HTYPE = constants.HTYPE_INSTANCE
6491 _OP_REQP = ["instance_name"]
6494 def CheckArguments(self):
6495 if not hasattr(self.op, 'nics'):
6497 if not hasattr(self.op, 'disks'):
6499 if not hasattr(self.op, 'beparams'):
6500 self.op.beparams = {}
6501 if not hasattr(self.op, 'hvparams'):
6502 self.op.hvparams = {}
6503 self.op.force = getattr(self.op, "force", False)
6504 if not (self.op.nics or self.op.disks or
6505 self.op.hvparams or self.op.beparams):
6506 raise errors.OpPrereqError("No changes submitted")
6510 for disk_op, disk_dict in self.op.disks:
6511 if disk_op == constants.DDM_REMOVE:
6514 elif disk_op == constants.DDM_ADD:
6517 if not isinstance(disk_op, int):
6518 raise errors.OpPrereqError("Invalid disk index")
6519 if not isinstance(disk_dict, dict):
6520 msg = "Invalid disk value: expected dict, got '%s'" % disk_dict
6521 raise errors.OpPrereqError(msg)
6523 if disk_op == constants.DDM_ADD:
6524 mode = disk_dict.setdefault('mode', constants.DISK_RDWR)
6525 if mode not in constants.DISK_ACCESS_SET:
6526 raise errors.OpPrereqError("Invalid disk access mode '%s'" % mode)
6527 size = disk_dict.get('size', None)
6529 raise errors.OpPrereqError("Required disk parameter size missing")
6532 except ValueError, err:
6533 raise errors.OpPrereqError("Invalid disk size parameter: %s" %
6535 disk_dict['size'] = size
6537 # modification of disk
6538 if 'size' in disk_dict:
6539 raise errors.OpPrereqError("Disk size change not possible, use"
6542 if disk_addremove > 1:
6543 raise errors.OpPrereqError("Only one disk add or remove operation"
6544 " supported at a time")
6548 for nic_op, nic_dict in self.op.nics:
6549 if nic_op == constants.DDM_REMOVE:
6552 elif nic_op == constants.DDM_ADD:
6555 if not isinstance(nic_op, int):
6556 raise errors.OpPrereqError("Invalid nic index")
6557 if not isinstance(nic_dict, dict):
6558 msg = "Invalid nic value: expected dict, got '%s'" % nic_dict
6559 raise errors.OpPrereqError(msg)
6561 # nic_dict should be a dict
6562 nic_ip = nic_dict.get('ip', None)
6563 if nic_ip is not None:
6564 if nic_ip.lower() == constants.VALUE_NONE:
6565 nic_dict['ip'] = None
6567 if not utils.IsValidIP(nic_ip):
6568 raise errors.OpPrereqError("Invalid IP address '%s'" % nic_ip)
6570 nic_bridge = nic_dict.get('bridge', None)
6571 nic_link = nic_dict.get('link', None)
6572 if nic_bridge and nic_link:
6573 raise errors.OpPrereqError("Cannot pass 'bridge' and 'link'"
6574 " at the same time")
6575 elif nic_bridge and nic_bridge.lower() == constants.VALUE_NONE:
6576 nic_dict['bridge'] = None
6577 elif nic_link and nic_link.lower() == constants.VALUE_NONE:
6578 nic_dict['link'] = None
6580 if nic_op == constants.DDM_ADD:
6581 nic_mac = nic_dict.get('mac', None)
6583 nic_dict['mac'] = constants.VALUE_AUTO
6585 if 'mac' in nic_dict:
6586 nic_mac = nic_dict['mac']
6587 if nic_mac not in (constants.VALUE_AUTO, constants.VALUE_GENERATE):
6588 if not utils.IsValidMac(nic_mac):
6589 raise errors.OpPrereqError("Invalid MAC address %s" % nic_mac)
6590 if nic_op != constants.DDM_ADD and nic_mac == constants.VALUE_AUTO:
6591 raise errors.OpPrereqError("'auto' is not a valid MAC address when"
6592 " modifying an existing nic")
6594 if nic_addremove > 1:
6595 raise errors.OpPrereqError("Only one NIC add or remove operation"
6596 " supported at a time")
6598 def ExpandNames(self):
6599 self._ExpandAndLockInstance()
6600 self.needed_locks[locking.LEVEL_NODE] = []
6601 self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
6603 def DeclareLocks(self, level):
6604 if level == locking.LEVEL_NODE:
6605 self._LockInstancesNodes()
6607 def BuildHooksEnv(self):
6610 This runs on the master, primary and secondaries.
6614 if constants.BE_MEMORY in self.be_new:
6615 args['memory'] = self.be_new[constants.BE_MEMORY]
6616 if constants.BE_VCPUS in self.be_new:
6617 args['vcpus'] = self.be_new[constants.BE_VCPUS]
6618 # TODO: export disk changes. Note: _BuildInstanceHookEnv* don't export disk
6619 # information at all.
6622 nic_override = dict(self.op.nics)
6623 c_nicparams = self.cluster.nicparams[constants.PP_DEFAULT]
6624 for idx, nic in enumerate(self.instance.nics):
6625 if idx in nic_override:
6626 this_nic_override = nic_override[idx]
6628 this_nic_override = {}
6629 if 'ip' in this_nic_override:
6630 ip = this_nic_override['ip']
6633 if 'mac' in this_nic_override:
6634 mac = this_nic_override['mac']
6637 if idx in self.nic_pnew:
6638 nicparams = self.nic_pnew[idx]
6640 nicparams = objects.FillDict(c_nicparams, nic.nicparams)
6641 mode = nicparams[constants.NIC_MODE]
6642 link = nicparams[constants.NIC_LINK]
6643 args['nics'].append((ip, mac, mode, link))
6644 if constants.DDM_ADD in nic_override:
6645 ip = nic_override[constants.DDM_ADD].get('ip', None)
6646 mac = nic_override[constants.DDM_ADD]['mac']
6647 nicparams = self.nic_pnew[constants.DDM_ADD]
6648 mode = nicparams[constants.NIC_MODE]
6649 link = nicparams[constants.NIC_LINK]
6650 args['nics'].append((ip, mac, mode, link))
6651 elif constants.DDM_REMOVE in nic_override:
6652 del args['nics'][-1]
6654 env = _BuildInstanceHookEnvByObject(self, self.instance, override=args)
6655 nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
6658 def _GetUpdatedParams(self, old_params, update_dict,
6659 default_values, parameter_types):
6660 """Return the new params dict for the given params.
6662 @type old_params: dict
6663 @param old_params: old parameters
6664 @type update_dict: dict
6665 @param update_dict: dict containing new parameter values,
6666 or constants.VALUE_DEFAULT to reset the
6667 parameter to its default value
6668 @type default_values: dict
6669 @param default_values: default values for the filled parameters
6670 @type parameter_types: dict
6671 @param parameter_types: dict mapping target dict keys to types
6672 in constants.ENFORCEABLE_TYPES
6673 @rtype: (dict, dict)
6674 @return: (new_parameters, filled_parameters)
6677 params_copy = copy.deepcopy(old_params)
6678 for key, val in update_dict.iteritems():
6679 if val == constants.VALUE_DEFAULT:
6681 del params_copy[key]
6685 params_copy[key] = val
6686 utils.ForceDictType(params_copy, parameter_types)
6687 params_filled = objects.FillDict(default_values, params_copy)
6688 return (params_copy, params_filled)
6690 def CheckPrereq(self):
6691 """Check prerequisites.
6693 This only checks the instance list against the existing names.
6696 self.force = self.op.force
6698 # checking the new params on the primary/secondary nodes
6700 instance = self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
6701 cluster = self.cluster = self.cfg.GetClusterInfo()
6702 assert self.instance is not None, \
6703 "Cannot retrieve locked instance %s" % self.op.instance_name
6704 pnode = instance.primary_node
6705 nodelist = list(instance.all_nodes)
6707 # hvparams processing
6708 if self.op.hvparams:
6709 i_hvdict, hv_new = self._GetUpdatedParams(
6710 instance.hvparams, self.op.hvparams,
6711 cluster.hvparams[instance.hypervisor],
6712 constants.HVS_PARAMETER_TYPES)
6714 hypervisor.GetHypervisor(
6715 instance.hypervisor).CheckParameterSyntax(hv_new)
6716 _CheckHVParams(self, nodelist, instance.hypervisor, hv_new)
6717 self.hv_new = hv_new # the new actual values
6718 self.hv_inst = i_hvdict # the new dict (without defaults)
6720 self.hv_new = self.hv_inst = {}
6722 # beparams processing
6723 if self.op.beparams:
6724 i_bedict, be_new = self._GetUpdatedParams(
6725 instance.beparams, self.op.beparams,
6726 cluster.beparams[constants.PP_DEFAULT],
6727 constants.BES_PARAMETER_TYPES)
6728 self.be_new = be_new # the new actual values
6729 self.be_inst = i_bedict # the new dict (without defaults)
6731 self.be_new = self.be_inst = {}
6735 if constants.BE_MEMORY in self.op.beparams and not self.force:
6736 mem_check_list = [pnode]
6737 if be_new[constants.BE_AUTO_BALANCE]:
6738 # either we changed auto_balance to yes or it was from before
6739 mem_check_list.extend(instance.secondary_nodes)
6740 instance_info = self.rpc.call_instance_info(pnode, instance.name,
6741 instance.hypervisor)
6742 nodeinfo = self.rpc.call_node_info(mem_check_list, self.cfg.GetVGName(),
6743 instance.hypervisor)
6744 pninfo = nodeinfo[pnode]
6745 msg = pninfo.fail_msg
6747 # Assume the primary node is unreachable and go ahead
6748 self.warn.append("Can't get info from primary node %s: %s" %
6750 elif not isinstance(pninfo.payload.get('memory_free', None), int):
6751 self.warn.append("Node data from primary node %s doesn't contain"
6752 " free memory information" % pnode)
6753 elif instance_info.fail_msg:
6754 self.warn.append("Can't get instance runtime information: %s" %
6755 instance_info.fail_msg)
6757 if instance_info.payload:
6758 current_mem = int(instance_info.payload['memory'])
6760 # Assume instance not running
6761 # (there is a slight race condition here, but it's not very probable,
6762 # and we have no other way to check)
6764 miss_mem = (be_new[constants.BE_MEMORY] - current_mem -
6765 pninfo.payload['memory_free'])
6767 raise errors.OpPrereqError("This change will prevent the instance"
6768 " from starting, due to %d MB of memory"
6769 " missing on its primary node" % miss_mem)
6771 if be_new[constants.BE_AUTO_BALANCE]:
6772 for node, nres in nodeinfo.items():
6773 if node not in instance.secondary_nodes:
6777 self.warn.append("Can't get info from secondary node %s: %s" %
6779 elif not isinstance(nres.payload.get('memory_free', None), int):
6780 self.warn.append("Secondary node %s didn't return free"
6781 " memory information" % node)
6782 elif be_new[constants.BE_MEMORY] > nres.payload['memory_free']:
6783 self.warn.append("Not enough memory to failover instance to"
6784 " secondary node %s" % node)
6789 for nic_op, nic_dict in self.op.nics:
6790 if nic_op == constants.DDM_REMOVE:
6791 if not instance.nics:
6792 raise errors.OpPrereqError("Instance has no NICs, cannot remove")
6794 if nic_op != constants.DDM_ADD:
6796 if nic_op < 0 or nic_op >= len(instance.nics):
6797 raise errors.OpPrereqError("Invalid NIC index %s, valid values"
6799 (nic_op, len(instance.nics)))
6800 old_nic_params = instance.nics[nic_op].nicparams
6801 old_nic_ip = instance.nics[nic_op].ip
6806 update_params_dict = dict([(key, nic_dict[key])
6807 for key in constants.NICS_PARAMETERS
6808 if key in nic_dict])
6810 if 'bridge' in nic_dict:
6811 update_params_dict[constants.NIC_LINK] = nic_dict['bridge']
6813 new_nic_params, new_filled_nic_params = \
6814 self._GetUpdatedParams(old_nic_params, update_params_dict,
6815 cluster.nicparams[constants.PP_DEFAULT],
6816 constants.NICS_PARAMETER_TYPES)
6817 objects.NIC.CheckParameterSyntax(new_filled_nic_params)
6818 self.nic_pinst[nic_op] = new_nic_params
6819 self.nic_pnew[nic_op] = new_filled_nic_params
6820 new_nic_mode = new_filled_nic_params[constants.NIC_MODE]
6822 if new_nic_mode == constants.NIC_MODE_BRIDGED:
6823 nic_bridge = new_filled_nic_params[constants.NIC_LINK]
6824 msg = self.rpc.call_bridges_exist(pnode, [nic_bridge]).fail_msg
6826 msg = "Error checking bridges on node %s: %s" % (pnode, msg)
6828 self.warn.append(msg)
6830 raise errors.OpPrereqError(msg)
6831 if new_nic_mode == constants.NIC_MODE_ROUTED:
6832 if 'ip' in nic_dict:
6833 nic_ip = nic_dict['ip']
6837 raise errors.OpPrereqError('Cannot set the nic ip to None'
6839 if 'mac' in nic_dict:
6840 nic_mac = nic_dict['mac']
6842 raise errors.OpPrereqError('Cannot set the nic mac to None')
6843 elif nic_mac in (constants.VALUE_AUTO, constants.VALUE_GENERATE):
6844 # otherwise generate the mac
6845 nic_dict['mac'] = self.cfg.GenerateMAC()
6847 # or validate/reserve the current one
6848 if self.cfg.IsMacInUse(nic_mac):
6849 raise errors.OpPrereqError("MAC address %s already in use"
6850 " in cluster" % nic_mac)
6853 if self.op.disks and instance.disk_template == constants.DT_DISKLESS:
6854 raise errors.OpPrereqError("Disk operations not supported for"
6855 " diskless instances")
6856 for disk_op, disk_dict in self.op.disks:
6857 if disk_op == constants.DDM_REMOVE:
6858 if len(instance.disks) == 1:
6859 raise errors.OpPrereqError("Cannot remove the last disk of"
6861 ins_l = self.rpc.call_instance_list([pnode], [instance.hypervisor])
6862 ins_l = ins_l[pnode]
6863 msg = ins_l.fail_msg
6865 raise errors.OpPrereqError("Can't contact node %s: %s" %
6867 if instance.name in ins_l.payload:
6868 raise errors.OpPrereqError("Instance is running, can't remove"
6871 if (disk_op == constants.DDM_ADD and
6872 len(instance.nics) >= constants.MAX_DISKS):
6873 raise errors.OpPrereqError("Instance has too many disks (%d), cannot"
6874 " add more" % constants.MAX_DISKS)
6875 if disk_op not in (constants.DDM_ADD, constants.DDM_REMOVE):
6877 if disk_op < 0 or disk_op >= len(instance.disks):
6878 raise errors.OpPrereqError("Invalid disk index %s, valid values"
6880 (disk_op, len(instance.disks)))
6884 def Exec(self, feedback_fn):
6885 """Modifies an instance.
6887 All parameters take effect only at the next restart of the instance.
6890 # Process here the warnings from CheckPrereq, as we don't have a
6891 # feedback_fn there.
6892 for warn in self.warn:
6893 feedback_fn("WARNING: %s" % warn)
6896 instance = self.instance
6897 cluster = self.cluster
6899 for disk_op, disk_dict in self.op.disks:
6900 if disk_op == constants.DDM_REMOVE:
6901 # remove the last disk
6902 device = instance.disks.pop()
6903 device_idx = len(instance.disks)
6904 for node, disk in device.ComputeNodeTree(instance.primary_node):
6905 self.cfg.SetDiskID(disk, node)
6906 msg = self.rpc.call_blockdev_remove(node, disk).fail_msg
6908 self.LogWarning("Could not remove disk/%d on node %s: %s,"
6909 " continuing anyway", device_idx, node, msg)
6910 result.append(("disk/%d" % device_idx, "remove"))
6911 elif disk_op == constants.DDM_ADD:
6913 if instance.disk_template == constants.DT_FILE:
6914 file_driver, file_path = instance.disks[0].logical_id
6915 file_path = os.path.dirname(file_path)
6917 file_driver = file_path = None
6918 disk_idx_base = len(instance.disks)
6919 new_disk = _GenerateDiskTemplate(self,
6920 instance.disk_template,
6921 instance.name, instance.primary_node,
6922 instance.secondary_nodes,
6927 instance.disks.append(new_disk)
6928 info = _GetInstanceInfoText(instance)
6930 logging.info("Creating volume %s for instance %s",
6931 new_disk.iv_name, instance.name)
6932 # Note: this needs to be kept in sync with _CreateDisks
6934 for node in instance.all_nodes:
6935 f_create = node == instance.primary_node
6937 _CreateBlockDev(self, node, instance, new_disk,
6938 f_create, info, f_create)
6939 except errors.OpExecError, err:
6940 self.LogWarning("Failed to create volume %s (%s) on"
6942 new_disk.iv_name, new_disk, node, err)
6943 result.append(("disk/%d" % disk_idx_base, "add:size=%s,mode=%s" %
6944 (new_disk.size, new_disk.mode)))
6946 # change a given disk
6947 instance.disks[disk_op].mode = disk_dict['mode']
6948 result.append(("disk.mode/%d" % disk_op, disk_dict['mode']))
6950 for nic_op, nic_dict in self.op.nics:
6951 if nic_op == constants.DDM_REMOVE:
6952 # remove the last nic
6953 del instance.nics[-1]
6954 result.append(("nic.%d" % len(instance.nics), "remove"))
6955 elif nic_op == constants.DDM_ADD:
6956 # mac and bridge should be set, by now
6957 mac = nic_dict['mac']
6958 ip = nic_dict.get('ip', None)
6959 nicparams = self.nic_pinst[constants.DDM_ADD]
6960 new_nic = objects.NIC(mac=mac, ip=ip, nicparams=nicparams)
6961 instance.nics.append(new_nic)
6962 result.append(("nic.%d" % (len(instance.nics) - 1),
6963 "add:mac=%s,ip=%s,mode=%s,link=%s" %
6964 (new_nic.mac, new_nic.ip,
6965 self.nic_pnew[constants.DDM_ADD][constants.NIC_MODE],
6966 self.nic_pnew[constants.DDM_ADD][constants.NIC_LINK]
6969 for key in 'mac', 'ip':
6971 setattr(instance.nics[nic_op], key, nic_dict[key])
6972 if nic_op in self.nic_pnew:
6973 instance.nics[nic_op].nicparams = self.nic_pnew[nic_op]
6974 for key, val in nic_dict.iteritems():
6975 result.append(("nic.%s/%d" % (key, nic_op), val))
6978 if self.op.hvparams:
6979 instance.hvparams = self.hv_inst
6980 for key, val in self.op.hvparams.iteritems():
6981 result.append(("hv/%s" % key, val))
6984 if self.op.beparams:
6985 instance.beparams = self.be_inst
6986 for key, val in self.op.beparams.iteritems():
6987 result.append(("be/%s" % key, val))
6989 self.cfg.Update(instance)
6994 class LUQueryExports(NoHooksLU):
6995 """Query the exports list
6998 _OP_REQP = ['nodes']
7001 def ExpandNames(self):
7002 self.needed_locks = {}
7003 self.share_locks[locking.LEVEL_NODE] = 1
7004 if not self.op.nodes:
7005 self.needed_locks[locking.LEVEL_NODE] = locking.ALL_SET
7007 self.needed_locks[locking.LEVEL_NODE] = \
7008 _GetWantedNodes(self, self.op.nodes)
7010 def CheckPrereq(self):
7011 """Check prerequisites.
7014 self.nodes = self.acquired_locks[locking.LEVEL_NODE]
7016 def Exec(self, feedback_fn):
7017 """Compute the list of all the exported system images.
7020 @return: a dictionary with the structure node->(export-list)
7021 where export-list is a list of the instances exported on
7025 rpcresult = self.rpc.call_export_list(self.nodes)
7027 for node in rpcresult:
7028 if rpcresult[node].fail_msg:
7029 result[node] = False
7031 result[node] = rpcresult[node].payload
7036 class LUExportInstance(LogicalUnit):
7037 """Export an instance to an image in the cluster.
7040 HPATH = "instance-export"
7041 HTYPE = constants.HTYPE_INSTANCE
7042 _OP_REQP = ["instance_name", "target_node", "shutdown"]
7045 def ExpandNames(self):
7046 self._ExpandAndLockInstance()
7047 # FIXME: lock only instance primary and destination node
7049 # Sad but true, for now we have do lock all nodes, as we don't know where
7050 # the previous export might be, and and in this LU we search for it and
7051 # remove it from its current node. In the future we could fix this by:
7052 # - making a tasklet to search (share-lock all), then create the new one,
7053 # then one to remove, after
7054 # - removing the removal operation altogether
7055 self.needed_locks[locking.LEVEL_NODE] = locking.ALL_SET
7057 def DeclareLocks(self, level):
7058 """Last minute lock declaration."""
7059 # All nodes are locked anyway, so nothing to do here.
7061 def BuildHooksEnv(self):
7064 This will run on the master, primary node and target node.
7068 "EXPORT_NODE": self.op.target_node,
7069 "EXPORT_DO_SHUTDOWN": self.op.shutdown,
7071 env.update(_BuildInstanceHookEnvByObject(self, self.instance))
7072 nl = [self.cfg.GetMasterNode(), self.instance.primary_node,
7073 self.op.target_node]
7076 def CheckPrereq(self):
7077 """Check prerequisites.
7079 This checks that the instance and node names are valid.
7082 instance_name = self.op.instance_name
7083 self.instance = self.cfg.GetInstanceInfo(instance_name)
7084 assert self.instance is not None, \
7085 "Cannot retrieve locked instance %s" % self.op.instance_name
7086 _CheckNodeOnline(self, self.instance.primary_node)
7088 self.dst_node = self.cfg.GetNodeInfo(
7089 self.cfg.ExpandNodeName(self.op.target_node))
7091 if self.dst_node is None:
7092 # This is wrong node name, not a non-locked node
7093 raise errors.OpPrereqError("Wrong node name %s" % self.op.target_node)
7094 _CheckNodeOnline(self, self.dst_node.name)
7095 _CheckNodeNotDrained(self, self.dst_node.name)
7097 # instance disk type verification
7098 for disk in self.instance.disks:
7099 if disk.dev_type == constants.LD_FILE:
7100 raise errors.OpPrereqError("Export not supported for instances with"
7101 " file-based disks")
7103 def Exec(self, feedback_fn):
7104 """Export an instance to an image in the cluster.
7107 instance = self.instance
7108 dst_node = self.dst_node
7109 src_node = instance.primary_node
7110 if self.op.shutdown:
7111 # shutdown the instance, but not the disks
7112 result = self.rpc.call_instance_shutdown(src_node, instance)
7113 result.Raise("Could not shutdown instance %s on"
7114 " node %s" % (instance.name, src_node))
7116 vgname = self.cfg.GetVGName()
7120 # set the disks ID correctly since call_instance_start needs the
7121 # correct drbd minor to create the symlinks
7122 for disk in instance.disks:
7123 self.cfg.SetDiskID(disk, src_node)
7126 for idx, disk in enumerate(instance.disks):
7127 # result.payload will be a snapshot of an lvm leaf of the one we passed
7128 result = self.rpc.call_blockdev_snapshot(src_node, disk)
7129 msg = result.fail_msg
7131 self.LogWarning("Could not snapshot disk/%s on node %s: %s",
7133 snap_disks.append(False)
7135 disk_id = (vgname, result.payload)
7136 new_dev = objects.Disk(dev_type=constants.LD_LV, size=disk.size,
7137 logical_id=disk_id, physical_id=disk_id,
7138 iv_name=disk.iv_name)
7139 snap_disks.append(new_dev)
7142 if self.op.shutdown and instance.admin_up:
7143 result = self.rpc.call_instance_start(src_node, instance, None, None)
7144 msg = result.fail_msg
7146 _ShutdownInstanceDisks(self, instance)
7147 raise errors.OpExecError("Could not start instance: %s" % msg)
7149 # TODO: check for size
7151 cluster_name = self.cfg.GetClusterName()
7152 for idx, dev in enumerate(snap_disks):
7154 result = self.rpc.call_snapshot_export(src_node, dev, dst_node.name,
7155 instance, cluster_name, idx)
7156 msg = result.fail_msg
7158 self.LogWarning("Could not export disk/%s from node %s to"
7159 " node %s: %s", idx, src_node, dst_node.name, msg)
7160 msg = self.rpc.call_blockdev_remove(src_node, dev).fail_msg
7162 self.LogWarning("Could not remove snapshot for disk/%d from node"
7163 " %s: %s", idx, src_node, msg)
7165 result = self.rpc.call_finalize_export(dst_node.name, instance, snap_disks)
7166 msg = result.fail_msg
7168 self.LogWarning("Could not finalize export for instance %s"
7169 " on node %s: %s", instance.name, dst_node.name, msg)
7171 nodelist = self.cfg.GetNodeList()
7172 nodelist.remove(dst_node.name)
7174 # on one-node clusters nodelist will be empty after the removal
7175 # if we proceed the backup would be removed because OpQueryExports
7176 # substitutes an empty list with the full cluster node list.
7177 iname = instance.name
7179 exportlist = self.rpc.call_export_list(nodelist)
7180 for node in exportlist:
7181 if exportlist[node].fail_msg:
7183 if iname in exportlist[node].payload:
7184 msg = self.rpc.call_export_remove(node, iname).fail_msg
7186 self.LogWarning("Could not remove older export for instance %s"
7187 " on node %s: %s", iname, node, msg)
7190 class LURemoveExport(NoHooksLU):
7191 """Remove exports related to the named instance.
7194 _OP_REQP = ["instance_name"]
7197 def ExpandNames(self):
7198 self.needed_locks = {}
7199 # We need all nodes to be locked in order for RemoveExport to work, but we
7200 # don't need to lock the instance itself, as nothing will happen to it (and
7201 # we can remove exports also for a removed instance)
7202 self.needed_locks[locking.LEVEL_NODE] = locking.ALL_SET
7204 def CheckPrereq(self):
7205 """Check prerequisites.
7209 def Exec(self, feedback_fn):
7210 """Remove any export.
7213 instance_name = self.cfg.ExpandInstanceName(self.op.instance_name)
7214 # If the instance was not found we'll try with the name that was passed in.
7215 # This will only work if it was an FQDN, though.
7217 if not instance_name:
7219 instance_name = self.op.instance_name
7221 locked_nodes = self.acquired_locks[locking.LEVEL_NODE]
7222 exportlist = self.rpc.call_export_list(locked_nodes)
7224 for node in exportlist:
7225 msg = exportlist[node].fail_msg
7227 self.LogWarning("Failed to query node %s (continuing): %s", node, msg)
7229 if instance_name in exportlist[node].payload:
7231 result = self.rpc.call_export_remove(node, instance_name)
7232 msg = result.fail_msg
7234 logging.error("Could not remove export for instance %s"
7235 " on node %s: %s", instance_name, node, msg)
7237 if fqdn_warn and not found:
7238 feedback_fn("Export not found. If trying to remove an export belonging"
7239 " to a deleted instance please use its Fully Qualified"
7243 class TagsLU(NoHooksLU):
7246 This is an abstract class which is the parent of all the other tags LUs.
7250 def ExpandNames(self):
7251 self.needed_locks = {}
7252 if self.op.kind == constants.TAG_NODE:
7253 name = self.cfg.ExpandNodeName(self.op.name)
7255 raise errors.OpPrereqError("Invalid node name (%s)" %
7258 self.needed_locks[locking.LEVEL_NODE] = name
7259 elif self.op.kind == constants.TAG_INSTANCE:
7260 name = self.cfg.ExpandInstanceName(self.op.name)
7262 raise errors.OpPrereqError("Invalid instance name (%s)" %
7265 self.needed_locks[locking.LEVEL_INSTANCE] = name
7267 def CheckPrereq(self):
7268 """Check prerequisites.
7271 if self.op.kind == constants.TAG_CLUSTER:
7272 self.target = self.cfg.GetClusterInfo()
7273 elif self.op.kind == constants.TAG_NODE:
7274 self.target = self.cfg.GetNodeInfo(self.op.name)
7275 elif self.op.kind == constants.TAG_INSTANCE:
7276 self.target = self.cfg.GetInstanceInfo(self.op.name)
7278 raise errors.OpPrereqError("Wrong tag type requested (%s)" %
7282 class LUGetTags(TagsLU):
7283 """Returns the tags of a given object.
7286 _OP_REQP = ["kind", "name"]
7289 def Exec(self, feedback_fn):
7290 """Returns the tag list.
7293 return list(self.target.GetTags())
7296 class LUSearchTags(NoHooksLU):
7297 """Searches the tags for a given pattern.
7300 _OP_REQP = ["pattern"]
7303 def ExpandNames(self):
7304 self.needed_locks = {}
7306 def CheckPrereq(self):
7307 """Check prerequisites.
7309 This checks the pattern passed for validity by compiling it.
7313 self.re = re.compile(self.op.pattern)
7314 except re.error, err:
7315 raise errors.OpPrereqError("Invalid search pattern '%s': %s" %
7316 (self.op.pattern, err))
7318 def Exec(self, feedback_fn):
7319 """Returns the tag list.
7323 tgts = [("/cluster", cfg.GetClusterInfo())]
7324 ilist = cfg.GetAllInstancesInfo().values()
7325 tgts.extend([("/instances/%s" % i.name, i) for i in ilist])
7326 nlist = cfg.GetAllNodesInfo().values()
7327 tgts.extend([("/nodes/%s" % n.name, n) for n in nlist])
7329 for path, target in tgts:
7330 for tag in target.GetTags():
7331 if self.re.search(tag):
7332 results.append((path, tag))
7336 class LUAddTags(TagsLU):
7337 """Sets a tag on a given object.
7340 _OP_REQP = ["kind", "name", "tags"]
7343 def CheckPrereq(self):
7344 """Check prerequisites.
7346 This checks the type and length of the tag name and value.
7349 TagsLU.CheckPrereq(self)
7350 for tag in self.op.tags:
7351 objects.TaggableObject.ValidateTag(tag)
7353 def Exec(self, feedback_fn):
7358 for tag in self.op.tags:
7359 self.target.AddTag(tag)
7360 except errors.TagError, err:
7361 raise errors.OpExecError("Error while setting tag: %s" % str(err))
7363 self.cfg.Update(self.target)
7364 except errors.ConfigurationError:
7365 raise errors.OpRetryError("There has been a modification to the"
7366 " config file and the operation has been"
7367 " aborted. Please retry.")
7370 class LUDelTags(TagsLU):
7371 """Delete a list of tags from a given object.
7374 _OP_REQP = ["kind", "name", "tags"]
7377 def CheckPrereq(self):
7378 """Check prerequisites.
7380 This checks that we have the given tag.
7383 TagsLU.CheckPrereq(self)
7384 for tag in self.op.tags:
7385 objects.TaggableObject.ValidateTag(tag)
7386 del_tags = frozenset(self.op.tags)
7387 cur_tags = self.target.GetTags()
7388 if not del_tags <= cur_tags:
7389 diff_tags = del_tags - cur_tags
7390 diff_names = ["'%s'" % tag for tag in diff_tags]
7392 raise errors.OpPrereqError("Tag(s) %s not found" %
7393 (",".join(diff_names)))
7395 def Exec(self, feedback_fn):
7396 """Remove the tag from the object.
7399 for tag in self.op.tags:
7400 self.target.RemoveTag(tag)
7402 self.cfg.Update(self.target)
7403 except errors.ConfigurationError:
7404 raise errors.OpRetryError("There has been a modification to the"
7405 " config file and the operation has been"
7406 " aborted. Please retry.")
7409 class LUTestDelay(NoHooksLU):
7410 """Sleep for a specified amount of time.
7412 This LU sleeps on the master and/or nodes for a specified amount of
7416 _OP_REQP = ["duration", "on_master", "on_nodes"]
7419 def ExpandNames(self):
7420 """Expand names and set required locks.
7422 This expands the node list, if any.
7425 self.needed_locks = {}
7426 if self.op.on_nodes:
7427 # _GetWantedNodes can be used here, but is not always appropriate to use
7428 # this way in ExpandNames. Check LogicalUnit.ExpandNames docstring for
7430 self.op.on_nodes = _GetWantedNodes(self, self.op.on_nodes)
7431 self.needed_locks[locking.LEVEL_NODE] = self.op.on_nodes
7433 def CheckPrereq(self):
7434 """Check prerequisites.
7438 def Exec(self, feedback_fn):
7439 """Do the actual sleep.
7442 if self.op.on_master:
7443 if not utils.TestDelay(self.op.duration):
7444 raise errors.OpExecError("Error during master delay test")
7445 if self.op.on_nodes:
7446 result = self.rpc.call_test_delay(self.op.on_nodes, self.op.duration)
7447 for node, node_result in result.items():
7448 node_result.Raise("Failure during rpc call to node %s" % node)
7451 class IAllocator(object):
7452 """IAllocator framework.
7454 An IAllocator instance has three sets of attributes:
7455 - cfg that is needed to query the cluster
7456 - input data (all members of the _KEYS class attribute are required)
7457 - four buffer attributes (in|out_data|text), that represent the
7458 input (to the external script) in text and data structure format,
7459 and the output from it, again in two formats
7460 - the result variables from the script (success, info, nodes) for
7465 "mem_size", "disks", "disk_template",
7466 "os", "tags", "nics", "vcpus", "hypervisor",
7472 def __init__(self, cfg, rpc, mode, name, **kwargs):
7475 # init buffer variables
7476 self.in_text = self.out_text = self.in_data = self.out_data = None
7477 # init all input fields so that pylint is happy
7480 self.mem_size = self.disks = self.disk_template = None
7481 self.os = self.tags = self.nics = self.vcpus = None
7482 self.hypervisor = None
7483 self.relocate_from = None
7485 self.required_nodes = None
7486 # init result fields
7487 self.success = self.info = self.nodes = None
7488 if self.mode == constants.IALLOCATOR_MODE_ALLOC:
7489 keyset = self._ALLO_KEYS
7490 elif self.mode == constants.IALLOCATOR_MODE_RELOC:
7491 keyset = self._RELO_KEYS
7493 raise errors.ProgrammerError("Unknown mode '%s' passed to the"
7494 " IAllocator" % self.mode)
7496 if key not in keyset:
7497 raise errors.ProgrammerError("Invalid input parameter '%s' to"
7498 " IAllocator" % key)
7499 setattr(self, key, kwargs[key])
7501 if key not in kwargs:
7502 raise errors.ProgrammerError("Missing input parameter '%s' to"
7503 " IAllocator" % key)
7504 self._BuildInputData()
7506 def _ComputeClusterData(self):
7507 """Compute the generic allocator input data.
7509 This is the data that is independent of the actual operation.
7513 cluster_info = cfg.GetClusterInfo()
7516 "version": constants.IALLOCATOR_VERSION,
7517 "cluster_name": cfg.GetClusterName(),
7518 "cluster_tags": list(cluster_info.GetTags()),
7519 "enabled_hypervisors": list(cluster_info.enabled_hypervisors),
7520 # we don't have job IDs
7522 iinfo = cfg.GetAllInstancesInfo().values()
7523 i_list = [(inst, cluster_info.FillBE(inst)) for inst in iinfo]
7527 node_list = cfg.GetNodeList()
7529 if self.mode == constants.IALLOCATOR_MODE_ALLOC:
7530 hypervisor_name = self.hypervisor
7531 elif self.mode == constants.IALLOCATOR_MODE_RELOC:
7532 hypervisor_name = cfg.GetInstanceInfo(self.name).hypervisor
7534 node_data = self.rpc.call_node_info(node_list, cfg.GetVGName(),
7537 self.rpc.call_all_instances_info(node_list,
7538 cluster_info.enabled_hypervisors)
7539 for nname, nresult in node_data.items():
7540 # first fill in static (config-based) values
7541 ninfo = cfg.GetNodeInfo(nname)
7543 "tags": list(ninfo.GetTags()),
7544 "primary_ip": ninfo.primary_ip,
7545 "secondary_ip": ninfo.secondary_ip,
7546 "offline": ninfo.offline,
7547 "drained": ninfo.drained,
7548 "master_candidate": ninfo.master_candidate,
7551 if not ninfo.offline:
7552 nresult.Raise("Can't get data for node %s" % nname)
7553 node_iinfo[nname].Raise("Can't get node instance info from node %s" %
7555 remote_info = nresult.payload
7556 for attr in ['memory_total', 'memory_free', 'memory_dom0',
7557 'vg_size', 'vg_free', 'cpu_total']:
7558 if attr not in remote_info:
7559 raise errors.OpExecError("Node '%s' didn't return attribute"
7560 " '%s'" % (nname, attr))
7561 if not isinstance(remote_info[attr], int):
7562 raise errors.OpExecError("Node '%s' returned invalid value"
7564 (nname, attr, remote_info[attr]))
7565 # compute memory used by primary instances
7566 i_p_mem = i_p_up_mem = 0
7567 for iinfo, beinfo in i_list:
7568 if iinfo.primary_node == nname:
7569 i_p_mem += beinfo[constants.BE_MEMORY]
7570 if iinfo.name not in node_iinfo[nname].payload:
7573 i_used_mem = int(node_iinfo[nname].payload[iinfo.name]['memory'])
7574 i_mem_diff = beinfo[constants.BE_MEMORY] - i_used_mem
7575 remote_info['memory_free'] -= max(0, i_mem_diff)
7578 i_p_up_mem += beinfo[constants.BE_MEMORY]
7580 # compute memory used by instances
7582 "total_memory": remote_info['memory_total'],
7583 "reserved_memory": remote_info['memory_dom0'],
7584 "free_memory": remote_info['memory_free'],
7585 "total_disk": remote_info['vg_size'],
7586 "free_disk": remote_info['vg_free'],
7587 "total_cpus": remote_info['cpu_total'],
7588 "i_pri_memory": i_p_mem,
7589 "i_pri_up_memory": i_p_up_mem,
7593 node_results[nname] = pnr
7594 data["nodes"] = node_results
7598 for iinfo, beinfo in i_list:
7600 for nic in iinfo.nics:
7601 filled_params = objects.FillDict(
7602 cluster_info.nicparams[constants.PP_DEFAULT],
7604 nic_dict = {"mac": nic.mac,
7606 "mode": filled_params[constants.NIC_MODE],
7607 "link": filled_params[constants.NIC_LINK],
7609 if filled_params[constants.NIC_MODE] == constants.NIC_MODE_BRIDGED:
7610 nic_dict["bridge"] = filled_params[constants.NIC_LINK]
7611 nic_data.append(nic_dict)
7613 "tags": list(iinfo.GetTags()),
7614 "admin_up": iinfo.admin_up,
7615 "vcpus": beinfo[constants.BE_VCPUS],
7616 "memory": beinfo[constants.BE_MEMORY],
7618 "nodes": [iinfo.primary_node] + list(iinfo.secondary_nodes),
7620 "disks": [{"size": dsk.size, "mode": dsk.mode} for dsk in iinfo.disks],
7621 "disk_template": iinfo.disk_template,
7622 "hypervisor": iinfo.hypervisor,
7624 pir["disk_space_total"] = _ComputeDiskSize(iinfo.disk_template,
7626 instance_data[iinfo.name] = pir
7628 data["instances"] = instance_data
7632 def _AddNewInstance(self):
7633 """Add new instance data to allocator structure.
7635 This in combination with _AllocatorGetClusterData will create the
7636 correct structure needed as input for the allocator.
7638 The checks for the completeness of the opcode must have already been
7644 disk_space = _ComputeDiskSize(self.disk_template, self.disks)
7646 if self.disk_template in constants.DTS_NET_MIRROR:
7647 self.required_nodes = 2
7649 self.required_nodes = 1
7653 "disk_template": self.disk_template,
7656 "vcpus": self.vcpus,
7657 "memory": self.mem_size,
7658 "disks": self.disks,
7659 "disk_space_total": disk_space,
7661 "required_nodes": self.required_nodes,
7663 data["request"] = request
7665 def _AddRelocateInstance(self):
7666 """Add relocate instance data to allocator structure.
7668 This in combination with _IAllocatorGetClusterData will create the
7669 correct structure needed as input for the allocator.
7671 The checks for the completeness of the opcode must have already been
7675 instance = self.cfg.GetInstanceInfo(self.name)
7676 if instance is None:
7677 raise errors.ProgrammerError("Unknown instance '%s' passed to"
7678 " IAllocator" % self.name)
7680 if instance.disk_template not in constants.DTS_NET_MIRROR:
7681 raise errors.OpPrereqError("Can't relocate non-mirrored instances")
7683 if len(instance.secondary_nodes) != 1:
7684 raise errors.OpPrereqError("Instance has not exactly one secondary node")
7686 self.required_nodes = 1
7687 disk_sizes = [{'size': disk.size} for disk in instance.disks]
7688 disk_space = _ComputeDiskSize(instance.disk_template, disk_sizes)
7693 "disk_space_total": disk_space,
7694 "required_nodes": self.required_nodes,
7695 "relocate_from": self.relocate_from,
7697 self.in_data["request"] = request
7699 def _BuildInputData(self):
7700 """Build input data structures.
7703 self._ComputeClusterData()
7705 if self.mode == constants.IALLOCATOR_MODE_ALLOC:
7706 self._AddNewInstance()
7708 self._AddRelocateInstance()
7710 self.in_text = serializer.Dump(self.in_data)
7712 def Run(self, name, validate=True, call_fn=None):
7713 """Run an instance allocator and return the results.
7717 call_fn = self.rpc.call_iallocator_runner
7719 result = call_fn(self.cfg.GetMasterNode(), name, self.in_text)
7720 result.Raise("Failure while running the iallocator script")
7722 self.out_text = result.payload
7724 self._ValidateResult()
7726 def _ValidateResult(self):
7727 """Process the allocator results.
7729 This will process and if successful save the result in
7730 self.out_data and the other parameters.
7734 rdict = serializer.Load(self.out_text)
7735 except Exception, err:
7736 raise errors.OpExecError("Can't parse iallocator results: %s" % str(err))
7738 if not isinstance(rdict, dict):
7739 raise errors.OpExecError("Can't parse iallocator results: not a dict")
7741 for key in "success", "info", "nodes":
7742 if key not in rdict:
7743 raise errors.OpExecError("Can't parse iallocator results:"
7744 " missing key '%s'" % key)
7745 setattr(self, key, rdict[key])
7747 if not isinstance(rdict["nodes"], list):
7748 raise errors.OpExecError("Can't parse iallocator results: 'nodes' key"
7750 self.out_data = rdict
7753 class LUTestAllocator(NoHooksLU):
7754 """Run allocator tests.
7756 This LU runs the allocator tests
7759 _OP_REQP = ["direction", "mode", "name"]
7761 def CheckPrereq(self):
7762 """Check prerequisites.
7764 This checks the opcode parameters depending on the director and mode test.
7767 if self.op.mode == constants.IALLOCATOR_MODE_ALLOC:
7768 for attr in ["name", "mem_size", "disks", "disk_template",
7769 "os", "tags", "nics", "vcpus"]:
7770 if not hasattr(self.op, attr):
7771 raise errors.OpPrereqError("Missing attribute '%s' on opcode input" %
7773 iname = self.cfg.ExpandInstanceName(self.op.name)
7774 if iname is not None:
7775 raise errors.OpPrereqError("Instance '%s' already in the cluster" %
7777 if not isinstance(self.op.nics, list):
7778 raise errors.OpPrereqError("Invalid parameter 'nics'")
7779 for row in self.op.nics:
7780 if (not isinstance(row, dict) or
7783 "bridge" not in row):
7784 raise errors.OpPrereqError("Invalid contents of the"
7785 " 'nics' parameter")
7786 if not isinstance(self.op.disks, list):
7787 raise errors.OpPrereqError("Invalid parameter 'disks'")
7788 for row in self.op.disks:
7789 if (not isinstance(row, dict) or
7790 "size" not in row or
7791 not isinstance(row["size"], int) or
7792 "mode" not in row or
7793 row["mode"] not in ['r', 'w']):
7794 raise errors.OpPrereqError("Invalid contents of the"
7795 " 'disks' parameter")
7796 if not hasattr(self.op, "hypervisor") or self.op.hypervisor is None:
7797 self.op.hypervisor = self.cfg.GetHypervisorType()
7798 elif self.op.mode == constants.IALLOCATOR_MODE_RELOC:
7799 if not hasattr(self.op, "name"):
7800 raise errors.OpPrereqError("Missing attribute 'name' on opcode input")
7801 fname = self.cfg.ExpandInstanceName(self.op.name)
7803 raise errors.OpPrereqError("Instance '%s' not found for relocation" %
7805 self.op.name = fname
7806 self.relocate_from = self.cfg.GetInstanceInfo(fname).secondary_nodes
7808 raise errors.OpPrereqError("Invalid test allocator mode '%s'" %
7811 if self.op.direction == constants.IALLOCATOR_DIR_OUT:
7812 if not hasattr(self.op, "allocator") or self.op.allocator is None:
7813 raise errors.OpPrereqError("Missing allocator name")
7814 elif self.op.direction != constants.IALLOCATOR_DIR_IN:
7815 raise errors.OpPrereqError("Wrong allocator test '%s'" %
7818 def Exec(self, feedback_fn):
7819 """Run the allocator test.
7822 if self.op.mode == constants.IALLOCATOR_MODE_ALLOC:
7823 ial = IAllocator(self.cfg, self.rpc,
7826 mem_size=self.op.mem_size,
7827 disks=self.op.disks,
7828 disk_template=self.op.disk_template,
7832 vcpus=self.op.vcpus,
7833 hypervisor=self.op.hypervisor,
7836 ial = IAllocator(self.cfg, self.rpc,
7839 relocate_from=list(self.relocate_from),
7842 if self.op.direction == constants.IALLOCATOR_DIR_IN:
7843 result = ial.in_text
7845 ial.Run(self.op.allocator, validate=False)
7846 result = ial.out_text