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=W0201
26 # W0201 since most LU attributes are defined in CheckPrereq or similar
37 from ganeti import ssh
38 from ganeti import utils
39 from ganeti import errors
40 from ganeti import hypervisor
41 from ganeti import locking
42 from ganeti import constants
43 from ganeti import objects
44 from ganeti import serializer
45 from ganeti import ssconf
48 class LogicalUnit(object):
49 """Logical Unit base class.
51 Subclasses must follow these rules:
52 - implement ExpandNames
53 - implement CheckPrereq (except when tasklets are used)
54 - implement Exec (except when tasklets are used)
55 - implement BuildHooksEnv
56 - redefine HPATH and HTYPE
57 - optionally redefine their run requirements:
58 REQ_BGL: the LU needs to hold the Big Ganeti Lock exclusively
60 Note that all commands require root permissions.
62 @ivar dry_run_result: the value (if any) that will be returned to the caller
63 in dry-run mode (signalled by opcode dry_run parameter)
71 def __init__(self, processor, op, context, rpc):
72 """Constructor for LogicalUnit.
74 This needs to be overridden in derived classes in order to check op
80 self.cfg = context.cfg
81 self.context = context
83 # Dicts used to declare locking needs to mcpu
84 self.needed_locks = None
85 self.acquired_locks = {}
86 self.share_locks = dict.fromkeys(locking.LEVELS, 0)
88 self.remove_locks = {}
89 # Used to force good behavior when calling helper functions
90 self.recalculate_locks = {}
93 self.LogWarning = processor.LogWarning # pylint: disable-msg=C0103
94 self.LogInfo = processor.LogInfo # pylint: disable-msg=C0103
95 self.LogStep = processor.LogStep # pylint: disable-msg=C0103
97 self.dry_run_result = None
98 # support for generic debug attribute
99 if (not hasattr(self.op, "debug_level") or
100 not isinstance(self.op.debug_level, int)):
101 self.op.debug_level = 0
106 for attr_name in self._OP_REQP:
107 attr_val = getattr(op, attr_name, None)
109 raise errors.OpPrereqError("Required parameter '%s' missing" %
110 attr_name, errors.ECODE_INVAL)
112 self.CheckArguments()
115 """Returns the SshRunner object
119 self.__ssh = ssh.SshRunner(self.cfg.GetClusterName())
122 ssh = property(fget=__GetSSH)
124 def CheckArguments(self):
125 """Check syntactic validity for the opcode arguments.
127 This method is for doing a simple syntactic check and ensure
128 validity of opcode parameters, without any cluster-related
129 checks. While the same can be accomplished in ExpandNames and/or
130 CheckPrereq, doing these separate is better because:
132 - ExpandNames is left as as purely a lock-related function
133 - CheckPrereq is run after we have acquired locks (and possible
136 The function is allowed to change the self.op attribute so that
137 later methods can no longer worry about missing parameters.
142 def ExpandNames(self):
143 """Expand names for this LU.
145 This method is called before starting to execute the opcode, and it should
146 update all the parameters of the opcode to their canonical form (e.g. a
147 short node name must be fully expanded after this method has successfully
148 completed). This way locking, hooks, logging, ecc. can work correctly.
150 LUs which implement this method must also populate the self.needed_locks
151 member, as a dict with lock levels as keys, and a list of needed lock names
154 - use an empty dict if you don't need any lock
155 - if you don't need any lock at a particular level omit that level
156 - don't put anything for the BGL level
157 - if you want all locks at a level use locking.ALL_SET as a value
159 If you need to share locks (rather than acquire them exclusively) at one
160 level you can modify self.share_locks, setting a true value (usually 1) for
161 that level. By default locks are not shared.
163 This function can also define a list of tasklets, which then will be
164 executed in order instead of the usual LU-level CheckPrereq and Exec
165 functions, if those are not defined by the LU.
169 # Acquire all nodes and one instance
170 self.needed_locks = {
171 locking.LEVEL_NODE: locking.ALL_SET,
172 locking.LEVEL_INSTANCE: ['instance1.example.tld'],
174 # Acquire just two nodes
175 self.needed_locks = {
176 locking.LEVEL_NODE: ['node1.example.tld', 'node2.example.tld'],
179 self.needed_locks = {} # No, you can't leave it to the default value None
182 # The implementation of this method is mandatory only if the new LU is
183 # concurrent, so that old LUs don't need to be changed all at the same
186 self.needed_locks = {} # Exclusive LUs don't need locks.
188 raise NotImplementedError
190 def DeclareLocks(self, level):
191 """Declare LU locking needs for a level
193 While most LUs can just declare their locking needs at ExpandNames time,
194 sometimes there's the need to calculate some locks after having acquired
195 the ones before. This function is called just before acquiring locks at a
196 particular level, but after acquiring the ones at lower levels, and permits
197 such calculations. It can be used to modify self.needed_locks, and by
198 default it does nothing.
200 This function is only called if you have something already set in
201 self.needed_locks for the level.
203 @param level: Locking level which is going to be locked
204 @type level: member of ganeti.locking.LEVELS
208 def CheckPrereq(self):
209 """Check prerequisites for this LU.
211 This method should check that the prerequisites for the execution
212 of this LU are fulfilled. It can do internode communication, but
213 it should be idempotent - no cluster or system changes are
216 The method should raise errors.OpPrereqError in case something is
217 not fulfilled. Its return value is ignored.
219 This method should also update all the parameters of the opcode to
220 their canonical form if it hasn't been done by ExpandNames before.
223 if self.tasklets is not None:
224 for (idx, tl) in enumerate(self.tasklets):
225 logging.debug("Checking prerequisites for tasklet %s/%s",
226 idx + 1, len(self.tasklets))
229 raise NotImplementedError
231 def Exec(self, feedback_fn):
234 This method should implement the actual work. It should raise
235 errors.OpExecError for failures that are somewhat dealt with in
239 if self.tasklets is not None:
240 for (idx, tl) in enumerate(self.tasklets):
241 logging.debug("Executing tasklet %s/%s", idx + 1, len(self.tasklets))
244 raise NotImplementedError
246 def BuildHooksEnv(self):
247 """Build hooks environment for this LU.
249 This method should return a three-node tuple consisting of: a dict
250 containing the environment that will be used for running the
251 specific hook for this LU, a list of node names on which the hook
252 should run before the execution, and a list of node names on which
253 the hook should run after the execution.
255 The keys of the dict must not have 'GANETI_' prefixed as this will
256 be handled in the hooks runner. Also note additional keys will be
257 added by the hooks runner. If the LU doesn't define any
258 environment, an empty dict (and not None) should be returned.
260 No nodes should be returned as an empty list (and not None).
262 Note that if the HPATH for a LU class is None, this function will
266 raise NotImplementedError
268 def HooksCallBack(self, phase, hook_results, feedback_fn, lu_result):
269 """Notify the LU about the results of its hooks.
271 This method is called every time a hooks phase is executed, and notifies
272 the Logical Unit about the hooks' result. The LU can then use it to alter
273 its result based on the hooks. By default the method does nothing and the
274 previous result is passed back unchanged but any LU can define it if it
275 wants to use the local cluster hook-scripts somehow.
277 @param phase: one of L{constants.HOOKS_PHASE_POST} or
278 L{constants.HOOKS_PHASE_PRE}; it denotes the hooks phase
279 @param hook_results: the results of the multi-node hooks rpc call
280 @param feedback_fn: function used send feedback back to the caller
281 @param lu_result: the previous Exec result this LU had, or None
283 @return: the new Exec result, based on the previous result
287 # API must be kept, thus we ignore the unused argument and could
288 # be a function warnings
289 # pylint: disable-msg=W0613,R0201
292 def _ExpandAndLockInstance(self):
293 """Helper function to expand and lock an instance.
295 Many LUs that work on an instance take its name in self.op.instance_name
296 and need to expand it and then declare the expanded name for locking. This
297 function does it, and then updates self.op.instance_name to the expanded
298 name. It also initializes needed_locks as a dict, if this hasn't been done
302 if self.needed_locks is None:
303 self.needed_locks = {}
305 assert locking.LEVEL_INSTANCE not in self.needed_locks, \
306 "_ExpandAndLockInstance called with instance-level locks set"
307 self.op.instance_name = _ExpandInstanceName(self.cfg,
308 self.op.instance_name)
309 self.needed_locks[locking.LEVEL_INSTANCE] = self.op.instance_name
311 def _LockInstancesNodes(self, primary_only=False):
312 """Helper function to declare instances' nodes for locking.
314 This function should be called after locking one or more instances to lock
315 their nodes. Its effect is populating self.needed_locks[locking.LEVEL_NODE]
316 with all primary or secondary nodes for instances already locked and
317 present in self.needed_locks[locking.LEVEL_INSTANCE].
319 It should be called from DeclareLocks, and for safety only works if
320 self.recalculate_locks[locking.LEVEL_NODE] is set.
322 In the future it may grow parameters to just lock some instance's nodes, or
323 to just lock primaries or secondary nodes, if needed.
325 If should be called in DeclareLocks in a way similar to::
327 if level == locking.LEVEL_NODE:
328 self._LockInstancesNodes()
330 @type primary_only: boolean
331 @param primary_only: only lock primary nodes of locked instances
334 assert locking.LEVEL_NODE in self.recalculate_locks, \
335 "_LockInstancesNodes helper function called with no nodes to recalculate"
337 # TODO: check if we're really been called with the instance locks held
339 # For now we'll replace self.needed_locks[locking.LEVEL_NODE], but in the
340 # future we might want to have different behaviors depending on the value
341 # of self.recalculate_locks[locking.LEVEL_NODE]
343 for instance_name in self.acquired_locks[locking.LEVEL_INSTANCE]:
344 instance = self.context.cfg.GetInstanceInfo(instance_name)
345 wanted_nodes.append(instance.primary_node)
347 wanted_nodes.extend(instance.secondary_nodes)
349 if self.recalculate_locks[locking.LEVEL_NODE] == constants.LOCKS_REPLACE:
350 self.needed_locks[locking.LEVEL_NODE] = wanted_nodes
351 elif self.recalculate_locks[locking.LEVEL_NODE] == constants.LOCKS_APPEND:
352 self.needed_locks[locking.LEVEL_NODE].extend(wanted_nodes)
354 del self.recalculate_locks[locking.LEVEL_NODE]
357 class NoHooksLU(LogicalUnit): # pylint: disable-msg=W0223
358 """Simple LU which runs no hooks.
360 This LU is intended as a parent for other LogicalUnits which will
361 run no hooks, in order to reduce duplicate code.
367 def BuildHooksEnv(self):
368 """Empty BuildHooksEnv for NoHooksLu.
370 This just raises an error.
373 assert False, "BuildHooksEnv called for NoHooksLUs"
377 """Tasklet base class.
379 Tasklets are subcomponents for LUs. LUs can consist entirely of tasklets or
380 they can mix legacy code with tasklets. Locking needs to be done in the LU,
381 tasklets know nothing about locks.
383 Subclasses must follow these rules:
384 - Implement CheckPrereq
388 def __init__(self, lu):
395 def CheckPrereq(self):
396 """Check prerequisites for this tasklets.
398 This method should check whether the prerequisites for the execution of
399 this tasklet are fulfilled. It can do internode communication, but it
400 should be idempotent - no cluster or system changes are allowed.
402 The method should raise errors.OpPrereqError in case something is not
403 fulfilled. Its return value is ignored.
405 This method should also update all parameters to their canonical form if it
406 hasn't been done before.
409 raise NotImplementedError
411 def Exec(self, feedback_fn):
412 """Execute the tasklet.
414 This method should implement the actual work. It should raise
415 errors.OpExecError for failures that are somewhat dealt with in code, or
419 raise NotImplementedError
422 def _GetWantedNodes(lu, nodes):
423 """Returns list of checked and expanded node names.
425 @type lu: L{LogicalUnit}
426 @param lu: the logical unit on whose behalf we execute
428 @param nodes: list of node names or None for all nodes
430 @return: the list of nodes, sorted
431 @raise errors.ProgrammerError: if the nodes parameter is wrong type
434 if not isinstance(nodes, list):
435 raise errors.OpPrereqError("Invalid argument type 'nodes'",
439 raise errors.ProgrammerError("_GetWantedNodes should only be called with a"
440 " non-empty list of nodes whose name is to be expanded.")
442 wanted = [_ExpandNodeName(lu.cfg, name) for name in nodes]
443 return utils.NiceSort(wanted)
446 def _GetWantedInstances(lu, instances):
447 """Returns list of checked and expanded instance names.
449 @type lu: L{LogicalUnit}
450 @param lu: the logical unit on whose behalf we execute
451 @type instances: list
452 @param instances: list of instance names or None for all instances
454 @return: the list of instances, sorted
455 @raise errors.OpPrereqError: if the instances parameter is wrong type
456 @raise errors.OpPrereqError: if any of the passed instances is not found
459 if not isinstance(instances, list):
460 raise errors.OpPrereqError("Invalid argument type 'instances'",
464 wanted = [_ExpandInstanceName(lu.cfg, name) for name in instances]
466 wanted = utils.NiceSort(lu.cfg.GetInstanceList())
470 def _CheckOutputFields(static, dynamic, selected):
471 """Checks whether all selected fields are valid.
473 @type static: L{utils.FieldSet}
474 @param static: static fields set
475 @type dynamic: L{utils.FieldSet}
476 @param dynamic: dynamic fields set
483 delta = f.NonMatching(selected)
485 raise errors.OpPrereqError("Unknown output fields selected: %s"
486 % ",".join(delta), errors.ECODE_INVAL)
489 def _CheckBooleanOpField(op, name):
490 """Validates boolean opcode parameters.
492 This will ensure that an opcode parameter is either a boolean value,
493 or None (but that it always exists).
496 val = getattr(op, name, None)
497 if not (val is None or isinstance(val, bool)):
498 raise errors.OpPrereqError("Invalid boolean parameter '%s' (%s)" %
499 (name, str(val)), errors.ECODE_INVAL)
500 setattr(op, name, val)
503 def _CheckGlobalHvParams(params):
504 """Validates that given hypervisor params are not global ones.
506 This will ensure that instances don't get customised versions of
510 used_globals = constants.HVC_GLOBALS.intersection(params)
512 msg = ("The following hypervisor parameters are global and cannot"
513 " be customized at instance level, please modify them at"
514 " cluster level: %s" % utils.CommaJoin(used_globals))
515 raise errors.OpPrereqError(msg, errors.ECODE_INVAL)
518 def _CheckNodeOnline(lu, node):
519 """Ensure that a given node is online.
521 @param lu: the LU on behalf of which we make the check
522 @param node: the node to check
523 @raise errors.OpPrereqError: if the node is offline
526 if lu.cfg.GetNodeInfo(node).offline:
527 raise errors.OpPrereqError("Can't use offline node %s" % node,
531 def _CheckNodeNotDrained(lu, node):
532 """Ensure that a given node is not drained.
534 @param lu: the LU on behalf of which we make the check
535 @param node: the node to check
536 @raise errors.OpPrereqError: if the node is drained
539 if lu.cfg.GetNodeInfo(node).drained:
540 raise errors.OpPrereqError("Can't use drained node %s" % node,
544 def _ExpandItemName(fn, name, kind):
545 """Expand an item name.
547 @param fn: the function to use for expansion
548 @param name: requested item name
549 @param kind: text description ('Node' or 'Instance')
550 @return: the resolved (full) name
551 @raise errors.OpPrereqError: if the item is not found
555 if full_name is None:
556 raise errors.OpPrereqError("%s '%s' not known" % (kind, name),
561 def _ExpandNodeName(cfg, name):
562 """Wrapper over L{_ExpandItemName} for nodes."""
563 return _ExpandItemName(cfg.ExpandNodeName, name, "Node")
566 def _ExpandInstanceName(cfg, name):
567 """Wrapper over L{_ExpandItemName} for instance."""
568 return _ExpandItemName(cfg.ExpandInstanceName, name, "Instance")
571 def _BuildInstanceHookEnv(name, primary_node, secondary_nodes, os_type, status,
572 memory, vcpus, nics, disk_template, disks,
573 bep, hvp, hypervisor_name):
574 """Builds instance related env variables for hooks
576 This builds the hook environment from individual variables.
579 @param name: the name of the instance
580 @type primary_node: string
581 @param primary_node: the name of the instance's primary node
582 @type secondary_nodes: list
583 @param secondary_nodes: list of secondary nodes as strings
584 @type os_type: string
585 @param os_type: the name of the instance's OS
586 @type status: boolean
587 @param status: the should_run status of the instance
589 @param memory: the memory size of the instance
591 @param vcpus: the count of VCPUs the instance has
593 @param nics: list of tuples (ip, mac, mode, link) representing
594 the NICs the instance has
595 @type disk_template: string
596 @param disk_template: the disk template of the instance
598 @param disks: the list of (size, mode) pairs
600 @param bep: the backend parameters for the instance
602 @param hvp: the hypervisor parameters for the instance
603 @type hypervisor_name: string
604 @param hypervisor_name: the hypervisor for the instance
606 @return: the hook environment for this instance
615 "INSTANCE_NAME": name,
616 "INSTANCE_PRIMARY": primary_node,
617 "INSTANCE_SECONDARIES": " ".join(secondary_nodes),
618 "INSTANCE_OS_TYPE": os_type,
619 "INSTANCE_STATUS": str_status,
620 "INSTANCE_MEMORY": memory,
621 "INSTANCE_VCPUS": vcpus,
622 "INSTANCE_DISK_TEMPLATE": disk_template,
623 "INSTANCE_HYPERVISOR": hypervisor_name,
627 nic_count = len(nics)
628 for idx, (ip, mac, mode, link) in enumerate(nics):
631 env["INSTANCE_NIC%d_IP" % idx] = ip
632 env["INSTANCE_NIC%d_MAC" % idx] = mac
633 env["INSTANCE_NIC%d_MODE" % idx] = mode
634 env["INSTANCE_NIC%d_LINK" % idx] = link
635 if mode == constants.NIC_MODE_BRIDGED:
636 env["INSTANCE_NIC%d_BRIDGE" % idx] = link
640 env["INSTANCE_NIC_COUNT"] = nic_count
643 disk_count = len(disks)
644 for idx, (size, mode) in enumerate(disks):
645 env["INSTANCE_DISK%d_SIZE" % idx] = size
646 env["INSTANCE_DISK%d_MODE" % idx] = mode
650 env["INSTANCE_DISK_COUNT"] = disk_count
652 for source, kind in [(bep, "BE"), (hvp, "HV")]:
653 for key, value in source.items():
654 env["INSTANCE_%s_%s" % (kind, key)] = value
659 def _NICListToTuple(lu, nics):
660 """Build a list of nic information tuples.
662 This list is suitable to be passed to _BuildInstanceHookEnv or as a return
663 value in LUQueryInstanceData.
665 @type lu: L{LogicalUnit}
666 @param lu: the logical unit on whose behalf we execute
667 @type nics: list of L{objects.NIC}
668 @param nics: list of nics to convert to hooks tuples
672 c_nicparams = lu.cfg.GetClusterInfo().nicparams[constants.PP_DEFAULT]
676 filled_params = objects.FillDict(c_nicparams, nic.nicparams)
677 mode = filled_params[constants.NIC_MODE]
678 link = filled_params[constants.NIC_LINK]
679 hooks_nics.append((ip, mac, mode, link))
683 def _BuildInstanceHookEnvByObject(lu, instance, override=None):
684 """Builds instance related env variables for hooks from an object.
686 @type lu: L{LogicalUnit}
687 @param lu: the logical unit on whose behalf we execute
688 @type instance: L{objects.Instance}
689 @param instance: the instance for which we should build the
692 @param override: dictionary with key/values that will override
695 @return: the hook environment dictionary
698 cluster = lu.cfg.GetClusterInfo()
699 bep = cluster.FillBE(instance)
700 hvp = cluster.FillHV(instance)
702 'name': instance.name,
703 'primary_node': instance.primary_node,
704 'secondary_nodes': instance.secondary_nodes,
705 'os_type': instance.os,
706 'status': instance.admin_up,
707 'memory': bep[constants.BE_MEMORY],
708 'vcpus': bep[constants.BE_VCPUS],
709 'nics': _NICListToTuple(lu, instance.nics),
710 'disk_template': instance.disk_template,
711 'disks': [(disk.size, disk.mode) for disk in instance.disks],
714 'hypervisor_name': instance.hypervisor,
717 args.update(override)
718 return _BuildInstanceHookEnv(**args) # pylint: disable-msg=W0142
721 def _AdjustCandidatePool(lu, exceptions):
722 """Adjust the candidate pool after node operations.
725 mod_list = lu.cfg.MaintainCandidatePool(exceptions)
727 lu.LogInfo("Promoted nodes to master candidate role: %s",
728 utils.CommaJoin(node.name for node in mod_list))
729 for name in mod_list:
730 lu.context.ReaddNode(name)
731 mc_now, mc_max, _ = lu.cfg.GetMasterCandidateStats(exceptions)
733 lu.LogInfo("Note: more nodes are candidates (%d) than desired (%d)" %
737 def _DecideSelfPromotion(lu, exceptions=None):
738 """Decide whether I should promote myself as a master candidate.
741 cp_size = lu.cfg.GetClusterInfo().candidate_pool_size
742 mc_now, mc_should, _ = lu.cfg.GetMasterCandidateStats(exceptions)
743 # the new node will increase mc_max with one, so:
744 mc_should = min(mc_should + 1, cp_size)
745 return mc_now < mc_should
748 def _CheckNicsBridgesExist(lu, target_nics, target_node,
749 profile=constants.PP_DEFAULT):
750 """Check that the brigdes needed by a list of nics exist.
753 c_nicparams = lu.cfg.GetClusterInfo().nicparams[profile]
754 paramslist = [objects.FillDict(c_nicparams, nic.nicparams)
755 for nic in target_nics]
756 brlist = [params[constants.NIC_LINK] for params in paramslist
757 if params[constants.NIC_MODE] == constants.NIC_MODE_BRIDGED]
759 result = lu.rpc.call_bridges_exist(target_node, brlist)
760 result.Raise("Error checking bridges on destination node '%s'" %
761 target_node, prereq=True, ecode=errors.ECODE_ENVIRON)
764 def _CheckInstanceBridgesExist(lu, instance, node=None):
765 """Check that the brigdes needed by an instance exist.
769 node = instance.primary_node
770 _CheckNicsBridgesExist(lu, instance.nics, node)
773 def _CheckOSVariant(os_obj, name):
774 """Check whether an OS name conforms to the os variants specification.
776 @type os_obj: L{objects.OS}
777 @param os_obj: OS object to check
779 @param name: OS name passed by the user, to check for validity
782 if not os_obj.supported_variants:
785 variant = name.split("+", 1)[1]
787 raise errors.OpPrereqError("OS name must include a variant",
790 if variant not in os_obj.supported_variants:
791 raise errors.OpPrereqError("Unsupported OS variant", errors.ECODE_INVAL)
794 def _GetNodeInstancesInner(cfg, fn):
795 return [i for i in cfg.GetAllInstancesInfo().values() if fn(i)]
798 def _GetNodeInstances(cfg, node_name):
799 """Returns a list of all primary and secondary instances on a node.
803 return _GetNodeInstancesInner(cfg, lambda inst: node_name in inst.all_nodes)
806 def _GetNodePrimaryInstances(cfg, node_name):
807 """Returns primary instances on a node.
810 return _GetNodeInstancesInner(cfg,
811 lambda inst: node_name == inst.primary_node)
814 def _GetNodeSecondaryInstances(cfg, node_name):
815 """Returns secondary instances on a node.
818 return _GetNodeInstancesInner(cfg,
819 lambda inst: node_name in inst.secondary_nodes)
822 def _GetStorageTypeArgs(cfg, storage_type):
823 """Returns the arguments for a storage type.
826 # Special case for file storage
827 if storage_type == constants.ST_FILE:
828 # storage.FileStorage wants a list of storage directories
829 return [[cfg.GetFileStorageDir()]]
834 def _FindFaultyInstanceDisks(cfg, rpc, instance, node_name, prereq):
837 for dev in instance.disks:
838 cfg.SetDiskID(dev, node_name)
840 result = rpc.call_blockdev_getmirrorstatus(node_name, instance.disks)
841 result.Raise("Failed to get disk status from node %s" % node_name,
842 prereq=prereq, ecode=errors.ECODE_ENVIRON)
844 for idx, bdev_status in enumerate(result.payload):
845 if bdev_status and bdev_status.ldisk_status == constants.LDS_FAULTY:
851 class LUPostInitCluster(LogicalUnit):
852 """Logical unit for running hooks after cluster initialization.
855 HPATH = "cluster-init"
856 HTYPE = constants.HTYPE_CLUSTER
859 def BuildHooksEnv(self):
863 env = {"OP_TARGET": self.cfg.GetClusterName()}
864 mn = self.cfg.GetMasterNode()
867 def CheckPrereq(self):
868 """No prerequisites to check.
873 def Exec(self, feedback_fn):
880 class LUDestroyCluster(LogicalUnit):
881 """Logical unit for destroying the cluster.
884 HPATH = "cluster-destroy"
885 HTYPE = constants.HTYPE_CLUSTER
888 def BuildHooksEnv(self):
892 env = {"OP_TARGET": self.cfg.GetClusterName()}
895 def CheckPrereq(self):
896 """Check prerequisites.
898 This checks whether the cluster is empty.
900 Any errors are signaled by raising errors.OpPrereqError.
903 master = self.cfg.GetMasterNode()
905 nodelist = self.cfg.GetNodeList()
906 if len(nodelist) != 1 or nodelist[0] != master:
907 raise errors.OpPrereqError("There are still %d node(s) in"
908 " this cluster." % (len(nodelist) - 1),
910 instancelist = self.cfg.GetInstanceList()
912 raise errors.OpPrereqError("There are still %d instance(s) in"
913 " this cluster." % len(instancelist),
916 def Exec(self, feedback_fn):
917 """Destroys the cluster.
920 master = self.cfg.GetMasterNode()
921 modify_ssh_setup = self.cfg.GetClusterInfo().modify_ssh_setup
923 # Run post hooks on master node before it's removed
924 hm = self.proc.hmclass(self.rpc.call_hooks_runner, self)
926 hm.RunPhase(constants.HOOKS_PHASE_POST, [master])
928 # pylint: disable-msg=W0702
929 self.LogWarning("Errors occurred running hooks on %s" % master)
931 result = self.rpc.call_node_stop_master(master, False)
932 result.Raise("Could not disable the master role")
935 priv_key, pub_key, _ = ssh.GetUserFiles(constants.GANETI_RUNAS)
936 utils.CreateBackup(priv_key)
937 utils.CreateBackup(pub_key)
942 class LUVerifyCluster(LogicalUnit):
943 """Verifies the cluster status.
946 HPATH = "cluster-verify"
947 HTYPE = constants.HTYPE_CLUSTER
948 _OP_REQP = ["skip_checks", "verbose", "error_codes", "debug_simulate_errors"]
953 TINSTANCE = "instance"
955 ECLUSTERCFG = (TCLUSTER, "ECLUSTERCFG")
956 EINSTANCEBADNODE = (TINSTANCE, "EINSTANCEBADNODE")
957 EINSTANCEDOWN = (TINSTANCE, "EINSTANCEDOWN")
958 EINSTANCELAYOUT = (TINSTANCE, "EINSTANCELAYOUT")
959 EINSTANCEMISSINGDISK = (TINSTANCE, "EINSTANCEMISSINGDISK")
960 EINSTANCEMISSINGDISK = (TINSTANCE, "EINSTANCEMISSINGDISK")
961 EINSTANCEWRONGNODE = (TINSTANCE, "EINSTANCEWRONGNODE")
962 ENODEDRBD = (TNODE, "ENODEDRBD")
963 ENODEFILECHECK = (TNODE, "ENODEFILECHECK")
964 ENODEHOOKS = (TNODE, "ENODEHOOKS")
965 ENODEHV = (TNODE, "ENODEHV")
966 ENODELVM = (TNODE, "ENODELVM")
967 ENODEN1 = (TNODE, "ENODEN1")
968 ENODENET = (TNODE, "ENODENET")
969 ENODEORPHANINSTANCE = (TNODE, "ENODEORPHANINSTANCE")
970 ENODEORPHANLV = (TNODE, "ENODEORPHANLV")
971 ENODERPC = (TNODE, "ENODERPC")
972 ENODESSH = (TNODE, "ENODESSH")
973 ENODEVERSION = (TNODE, "ENODEVERSION")
974 ENODESETUP = (TNODE, "ENODESETUP")
975 ENODETIME = (TNODE, "ENODETIME")
978 ETYPE_ERROR = "ERROR"
979 ETYPE_WARNING = "WARNING"
981 def ExpandNames(self):
982 self.needed_locks = {
983 locking.LEVEL_NODE: locking.ALL_SET,
984 locking.LEVEL_INSTANCE: locking.ALL_SET,
986 self.share_locks = dict.fromkeys(locking.LEVELS, 1)
988 def _Error(self, ecode, item, msg, *args, **kwargs):
989 """Format an error message.
991 Based on the opcode's error_codes parameter, either format a
992 parseable error code, or a simpler error string.
994 This must be called only from Exec and functions called from Exec.
997 ltype = kwargs.get(self.ETYPE_FIELD, self.ETYPE_ERROR)
999 # first complete the msg
1002 # then format the whole message
1003 if self.op.error_codes:
1004 msg = "%s:%s:%s:%s:%s" % (ltype, etxt, itype, item, msg)
1010 msg = "%s: %s%s: %s" % (ltype, itype, item, msg)
1011 # and finally report it via the feedback_fn
1012 self._feedback_fn(" - %s" % msg)
1014 def _ErrorIf(self, cond, *args, **kwargs):
1015 """Log an error message if the passed condition is True.
1018 cond = bool(cond) or self.op.debug_simulate_errors
1020 self._Error(*args, **kwargs)
1021 # do not mark the operation as failed for WARN cases only
1022 if kwargs.get(self.ETYPE_FIELD, self.ETYPE_ERROR) == self.ETYPE_ERROR:
1023 self.bad = self.bad or cond
1025 def _VerifyNode(self, nodeinfo, file_list, local_cksum,
1026 node_result, master_files, drbd_map, vg_name):
1027 """Run multiple tests against a node.
1031 - compares ganeti version
1032 - checks vg existence and size > 20G
1033 - checks config file checksum
1034 - checks ssh to other nodes
1036 @type nodeinfo: L{objects.Node}
1037 @param nodeinfo: the node to check
1038 @param file_list: required list of files
1039 @param local_cksum: dictionary of local files and their checksums
1040 @param node_result: the results from the node
1041 @param master_files: list of files that only masters should have
1042 @param drbd_map: the useddrbd minors for this node, in
1043 form of minor: (instance, must_exist) which correspond to instances
1044 and their running status
1045 @param vg_name: Ganeti Volume Group (result of self.cfg.GetVGName())
1048 node = nodeinfo.name
1049 _ErrorIf = self._ErrorIf # pylint: disable-msg=C0103
1051 # main result, node_result should be a non-empty dict
1052 test = not node_result or not isinstance(node_result, dict)
1053 _ErrorIf(test, self.ENODERPC, node,
1054 "unable to verify node: no data returned")
1058 # compares ganeti version
1059 local_version = constants.PROTOCOL_VERSION
1060 remote_version = node_result.get('version', None)
1061 test = not (remote_version and
1062 isinstance(remote_version, (list, tuple)) and
1063 len(remote_version) == 2)
1064 _ErrorIf(test, self.ENODERPC, node,
1065 "connection to node returned invalid data")
1069 test = local_version != remote_version[0]
1070 _ErrorIf(test, self.ENODEVERSION, node,
1071 "incompatible protocol versions: master %s,"
1072 " node %s", local_version, remote_version[0])
1076 # node seems compatible, we can actually try to look into its results
1078 # full package version
1079 self._ErrorIf(constants.RELEASE_VERSION != remote_version[1],
1080 self.ENODEVERSION, node,
1081 "software version mismatch: master %s, node %s",
1082 constants.RELEASE_VERSION, remote_version[1],
1083 code=self.ETYPE_WARNING)
1085 # checks vg existence and size > 20G
1086 if vg_name is not None:
1087 vglist = node_result.get(constants.NV_VGLIST, None)
1089 _ErrorIf(test, self.ENODELVM, node, "unable to check volume groups")
1091 vgstatus = utils.CheckVolumeGroupSize(vglist, vg_name,
1092 constants.MIN_VG_SIZE)
1093 _ErrorIf(vgstatus, self.ENODELVM, node, vgstatus)
1095 # checks config file checksum
1097 remote_cksum = node_result.get(constants.NV_FILELIST, None)
1098 test = not isinstance(remote_cksum, dict)
1099 _ErrorIf(test, self.ENODEFILECHECK, node,
1100 "node hasn't returned file checksum data")
1102 for file_name in file_list:
1103 node_is_mc = nodeinfo.master_candidate
1104 must_have = (file_name not in master_files) or node_is_mc
1106 test1 = file_name not in remote_cksum
1108 test2 = not test1 and remote_cksum[file_name] != local_cksum[file_name]
1110 test3 = not test1 and remote_cksum[file_name] == local_cksum[file_name]
1111 _ErrorIf(test1 and must_have, self.ENODEFILECHECK, node,
1112 "file '%s' missing", file_name)
1113 _ErrorIf(test2 and must_have, self.ENODEFILECHECK, node,
1114 "file '%s' has wrong checksum", file_name)
1115 # not candidate and this is not a must-have file
1116 _ErrorIf(test2 and not must_have, self.ENODEFILECHECK, node,
1117 "file '%s' should not exist on non master"
1118 " candidates (and the file is outdated)", file_name)
1119 # all good, except non-master/non-must have combination
1120 _ErrorIf(test3 and not must_have, self.ENODEFILECHECK, node,
1121 "file '%s' should not exist"
1122 " on non master candidates", file_name)
1126 test = constants.NV_NODELIST not in node_result
1127 _ErrorIf(test, self.ENODESSH, node,
1128 "node hasn't returned node ssh connectivity data")
1130 if node_result[constants.NV_NODELIST]:
1131 for a_node, a_msg in node_result[constants.NV_NODELIST].items():
1132 _ErrorIf(True, self.ENODESSH, node,
1133 "ssh communication with node '%s': %s", a_node, a_msg)
1135 test = constants.NV_NODENETTEST not in node_result
1136 _ErrorIf(test, self.ENODENET, node,
1137 "node hasn't returned node tcp connectivity data")
1139 if node_result[constants.NV_NODENETTEST]:
1140 nlist = utils.NiceSort(node_result[constants.NV_NODENETTEST].keys())
1142 _ErrorIf(True, self.ENODENET, node,
1143 "tcp communication with node '%s': %s",
1144 anode, node_result[constants.NV_NODENETTEST][anode])
1146 hyp_result = node_result.get(constants.NV_HYPERVISOR, None)
1147 if isinstance(hyp_result, dict):
1148 for hv_name, hv_result in hyp_result.iteritems():
1149 test = hv_result is not None
1150 _ErrorIf(test, self.ENODEHV, node,
1151 "hypervisor %s verify failure: '%s'", hv_name, hv_result)
1153 # check used drbd list
1154 if vg_name is not None:
1155 used_minors = node_result.get(constants.NV_DRBDLIST, [])
1156 test = not isinstance(used_minors, (tuple, list))
1157 _ErrorIf(test, self.ENODEDRBD, node,
1158 "cannot parse drbd status file: %s", str(used_minors))
1160 for minor, (iname, must_exist) in drbd_map.items():
1161 test = minor not in used_minors and must_exist
1162 _ErrorIf(test, self.ENODEDRBD, node,
1163 "drbd minor %d of instance %s is not active",
1165 for minor in used_minors:
1166 test = minor not in drbd_map
1167 _ErrorIf(test, self.ENODEDRBD, node,
1168 "unallocated drbd minor %d is in use", minor)
1169 test = node_result.get(constants.NV_NODESETUP,
1170 ["Missing NODESETUP results"])
1171 _ErrorIf(test, self.ENODESETUP, node, "node setup error: %s",
1175 if vg_name is not None:
1176 pvlist = node_result.get(constants.NV_PVLIST, None)
1177 test = pvlist is None
1178 _ErrorIf(test, self.ENODELVM, node, "Can't get PV list from node")
1180 # check that ':' is not present in PV names, since it's a
1181 # special character for lvcreate (denotes the range of PEs to
1183 for _, pvname, owner_vg in pvlist:
1184 test = ":" in pvname
1185 _ErrorIf(test, self.ENODELVM, node, "Invalid character ':' in PV"
1186 " '%s' of VG '%s'", pvname, owner_vg)
1188 def _VerifyInstance(self, instance, instanceconfig, node_vol_is,
1189 node_instance, n_offline):
1190 """Verify an instance.
1192 This function checks to see if the required block devices are
1193 available on the instance's node.
1196 _ErrorIf = self._ErrorIf # pylint: disable-msg=C0103
1197 node_current = instanceconfig.primary_node
1199 node_vol_should = {}
1200 instanceconfig.MapLVsByNode(node_vol_should)
1202 for node in node_vol_should:
1203 if node in n_offline:
1204 # ignore missing volumes on offline nodes
1206 for volume in node_vol_should[node]:
1207 test = node not in node_vol_is or volume not in node_vol_is[node]
1208 _ErrorIf(test, self.EINSTANCEMISSINGDISK, instance,
1209 "volume %s missing on node %s", volume, node)
1211 if instanceconfig.admin_up:
1212 test = ((node_current not in node_instance or
1213 not instance in node_instance[node_current]) and
1214 node_current not in n_offline)
1215 _ErrorIf(test, self.EINSTANCEDOWN, instance,
1216 "instance not running on its primary node %s",
1219 for node in node_instance:
1220 if (not node == node_current):
1221 test = instance in node_instance[node]
1222 _ErrorIf(test, self.EINSTANCEWRONGNODE, instance,
1223 "instance should not run on node %s", node)
1225 def _VerifyOrphanVolumes(self, node_vol_should, node_vol_is):
1226 """Verify if there are any unknown volumes in the cluster.
1228 The .os, .swap and backup volumes are ignored. All other volumes are
1229 reported as unknown.
1232 for node in node_vol_is:
1233 for volume in node_vol_is[node]:
1234 test = (node not in node_vol_should or
1235 volume not in node_vol_should[node])
1236 self._ErrorIf(test, self.ENODEORPHANLV, node,
1237 "volume %s is unknown", volume)
1239 def _VerifyOrphanInstances(self, instancelist, node_instance):
1240 """Verify the list of running instances.
1242 This checks what instances are running but unknown to the cluster.
1245 for node in node_instance:
1246 for o_inst in node_instance[node]:
1247 test = o_inst not in instancelist
1248 self._ErrorIf(test, self.ENODEORPHANINSTANCE, node,
1249 "instance %s on node %s should not exist", o_inst, node)
1251 def _VerifyNPlusOneMemory(self, node_info, instance_cfg):
1252 """Verify N+1 Memory Resilience.
1254 Check that if one single node dies we can still start all the instances it
1258 for node, nodeinfo in node_info.iteritems():
1259 # This code checks that every node which is now listed as secondary has
1260 # enough memory to host all instances it is supposed to should a single
1261 # other node in the cluster fail.
1262 # FIXME: not ready for failover to an arbitrary node
1263 # FIXME: does not support file-backed instances
1264 # WARNING: we currently take into account down instances as well as up
1265 # ones, considering that even if they're down someone might want to start
1266 # them even in the event of a node failure.
1267 for prinode, instances in nodeinfo['sinst-by-pnode'].iteritems():
1269 for instance in instances:
1270 bep = self.cfg.GetClusterInfo().FillBE(instance_cfg[instance])
1271 if bep[constants.BE_AUTO_BALANCE]:
1272 needed_mem += bep[constants.BE_MEMORY]
1273 test = nodeinfo['mfree'] < needed_mem
1274 self._ErrorIf(test, self.ENODEN1, node,
1275 "not enough memory on to accommodate"
1276 " failovers should peer node %s fail", prinode)
1278 def CheckPrereq(self):
1279 """Check prerequisites.
1281 Transform the list of checks we're going to skip into a set and check that
1282 all its members are valid.
1285 self.skip_set = frozenset(self.op.skip_checks)
1286 if not constants.VERIFY_OPTIONAL_CHECKS.issuperset(self.skip_set):
1287 raise errors.OpPrereqError("Invalid checks to be skipped specified",
1290 def BuildHooksEnv(self):
1293 Cluster-Verify hooks just ran in the post phase and their failure makes
1294 the output be logged in the verify output and the verification to fail.
1297 all_nodes = self.cfg.GetNodeList()
1299 "CLUSTER_TAGS": " ".join(self.cfg.GetClusterInfo().GetTags())
1301 for node in self.cfg.GetAllNodesInfo().values():
1302 env["NODE_TAGS_%s" % node.name] = " ".join(node.GetTags())
1304 return env, [], all_nodes
1306 def Exec(self, feedback_fn):
1307 """Verify integrity of cluster, performing various test on nodes.
1311 _ErrorIf = self._ErrorIf # pylint: disable-msg=C0103
1312 verbose = self.op.verbose
1313 self._feedback_fn = feedback_fn
1314 feedback_fn("* Verifying global settings")
1315 for msg in self.cfg.VerifyConfig():
1316 _ErrorIf(True, self.ECLUSTERCFG, None, msg)
1318 vg_name = self.cfg.GetVGName()
1319 hypervisors = self.cfg.GetClusterInfo().enabled_hypervisors
1320 nodelist = utils.NiceSort(self.cfg.GetNodeList())
1321 nodeinfo = [self.cfg.GetNodeInfo(nname) for nname in nodelist]
1322 instancelist = utils.NiceSort(self.cfg.GetInstanceList())
1323 instanceinfo = dict((iname, self.cfg.GetInstanceInfo(iname))
1324 for iname in instancelist)
1325 i_non_redundant = [] # Non redundant instances
1326 i_non_a_balanced = [] # Non auto-balanced instances
1327 n_offline = [] # List of offline nodes
1328 n_drained = [] # List of nodes being drained
1334 # FIXME: verify OS list
1335 # do local checksums
1336 master_files = [constants.CLUSTER_CONF_FILE]
1338 file_names = ssconf.SimpleStore().GetFileList()
1339 file_names.append(constants.SSL_CERT_FILE)
1340 file_names.append(constants.RAPI_CERT_FILE)
1341 file_names.extend(master_files)
1343 local_checksums = utils.FingerprintFiles(file_names)
1345 feedback_fn("* Gathering data (%d nodes)" % len(nodelist))
1346 node_verify_param = {
1347 constants.NV_FILELIST: file_names,
1348 constants.NV_NODELIST: [node.name for node in nodeinfo
1349 if not node.offline],
1350 constants.NV_HYPERVISOR: hypervisors,
1351 constants.NV_NODENETTEST: [(node.name, node.primary_ip,
1352 node.secondary_ip) for node in nodeinfo
1353 if not node.offline],
1354 constants.NV_INSTANCELIST: hypervisors,
1355 constants.NV_VERSION: None,
1356 constants.NV_HVINFO: self.cfg.GetHypervisorType(),
1357 constants.NV_NODESETUP: None,
1358 constants.NV_TIME: None,
1361 if vg_name is not None:
1362 node_verify_param[constants.NV_VGLIST] = None
1363 node_verify_param[constants.NV_LVLIST] = vg_name
1364 node_verify_param[constants.NV_PVLIST] = [vg_name]
1365 node_verify_param[constants.NV_DRBDLIST] = None
1367 # Due to the way our RPC system works, exact response times cannot be
1368 # guaranteed (e.g. a broken node could run into a timeout). By keeping the
1369 # time before and after executing the request, we can at least have a time
1371 nvinfo_starttime = time.time()
1372 all_nvinfo = self.rpc.call_node_verify(nodelist, node_verify_param,
1373 self.cfg.GetClusterName())
1374 nvinfo_endtime = time.time()
1376 cluster = self.cfg.GetClusterInfo()
1377 master_node = self.cfg.GetMasterNode()
1378 all_drbd_map = self.cfg.ComputeDRBDMap()
1380 feedback_fn("* Verifying node status")
1381 for node_i in nodeinfo:
1386 feedback_fn("* Skipping offline node %s" % (node,))
1387 n_offline.append(node)
1390 if node == master_node:
1392 elif node_i.master_candidate:
1393 ntype = "master candidate"
1394 elif node_i.drained:
1396 n_drained.append(node)
1400 feedback_fn("* Verifying node %s (%s)" % (node, ntype))
1402 msg = all_nvinfo[node].fail_msg
1403 _ErrorIf(msg, self.ENODERPC, node, "while contacting node: %s", msg)
1407 nresult = all_nvinfo[node].payload
1409 for minor, instance in all_drbd_map[node].items():
1410 test = instance not in instanceinfo
1411 _ErrorIf(test, self.ECLUSTERCFG, None,
1412 "ghost instance '%s' in temporary DRBD map", instance)
1413 # ghost instance should not be running, but otherwise we
1414 # don't give double warnings (both ghost instance and
1415 # unallocated minor in use)
1417 node_drbd[minor] = (instance, False)
1419 instance = instanceinfo[instance]
1420 node_drbd[minor] = (instance.name, instance.admin_up)
1422 self._VerifyNode(node_i, file_names, local_checksums,
1423 nresult, master_files, node_drbd, vg_name)
1425 lvdata = nresult.get(constants.NV_LVLIST, "Missing LV data")
1427 node_volume[node] = {}
1428 elif isinstance(lvdata, basestring):
1429 _ErrorIf(True, self.ENODELVM, node, "LVM problem on node: %s",
1430 utils.SafeEncode(lvdata))
1431 node_volume[node] = {}
1432 elif not isinstance(lvdata, dict):
1433 _ErrorIf(True, self.ENODELVM, node, "rpc call to node failed (lvlist)")
1436 node_volume[node] = lvdata
1439 idata = nresult.get(constants.NV_INSTANCELIST, None)
1440 test = not isinstance(idata, list)
1441 _ErrorIf(test, self.ENODEHV, node,
1442 "rpc call to node failed (instancelist): %s",
1443 utils.SafeEncode(str(idata)))
1447 node_instance[node] = idata
1450 nodeinfo = nresult.get(constants.NV_HVINFO, None)
1451 test = not isinstance(nodeinfo, dict)
1452 _ErrorIf(test, self.ENODEHV, node, "rpc call to node failed (hvinfo)")
1457 ntime = nresult.get(constants.NV_TIME, None)
1459 ntime_merged = utils.MergeTime(ntime)
1460 except (ValueError, TypeError):
1461 _ErrorIf(True, self.ENODETIME, node, "Node returned invalid time")
1463 if ntime_merged < (nvinfo_starttime - constants.NODE_MAX_CLOCK_SKEW):
1464 ntime_diff = "%.01fs" % abs(nvinfo_starttime - ntime_merged)
1465 elif ntime_merged > (nvinfo_endtime + constants.NODE_MAX_CLOCK_SKEW):
1466 ntime_diff = "%.01fs" % abs(ntime_merged - nvinfo_endtime)
1470 _ErrorIf(ntime_diff is not None, self.ENODETIME, node,
1471 "Node time diverges by at least %s from master node time",
1474 if ntime_diff is not None:
1479 "mfree": int(nodeinfo['memory_free']),
1482 # dictionary holding all instances this node is secondary for,
1483 # grouped by their primary node. Each key is a cluster node, and each
1484 # value is a list of instances which have the key as primary and the
1485 # current node as secondary. this is handy to calculate N+1 memory
1486 # availability if you can only failover from a primary to its
1488 "sinst-by-pnode": {},
1490 # FIXME: devise a free space model for file based instances as well
1491 if vg_name is not None:
1492 test = (constants.NV_VGLIST not in nresult or
1493 vg_name not in nresult[constants.NV_VGLIST])
1494 _ErrorIf(test, self.ENODELVM, node,
1495 "node didn't return data for the volume group '%s'"
1496 " - it is either missing or broken", vg_name)
1499 node_info[node]["dfree"] = int(nresult[constants.NV_VGLIST][vg_name])
1500 except (ValueError, KeyError):
1501 _ErrorIf(True, self.ENODERPC, node,
1502 "node returned invalid nodeinfo, check lvm/hypervisor")
1505 node_vol_should = {}
1507 feedback_fn("* Verifying instance status")
1508 for instance in instancelist:
1510 feedback_fn("* Verifying instance %s" % instance)
1511 inst_config = instanceinfo[instance]
1512 self._VerifyInstance(instance, inst_config, node_volume,
1513 node_instance, n_offline)
1514 inst_nodes_offline = []
1516 inst_config.MapLVsByNode(node_vol_should)
1518 instance_cfg[instance] = inst_config
1520 pnode = inst_config.primary_node
1521 _ErrorIf(pnode not in node_info and pnode not in n_offline,
1522 self.ENODERPC, pnode, "instance %s, connection to"
1523 " primary node failed", instance)
1524 if pnode in node_info:
1525 node_info[pnode]['pinst'].append(instance)
1527 if pnode in n_offline:
1528 inst_nodes_offline.append(pnode)
1530 # If the instance is non-redundant we cannot survive losing its primary
1531 # node, so we are not N+1 compliant. On the other hand we have no disk
1532 # templates with more than one secondary so that situation is not well
1534 # FIXME: does not support file-backed instances
1535 if len(inst_config.secondary_nodes) == 0:
1536 i_non_redundant.append(instance)
1537 _ErrorIf(len(inst_config.secondary_nodes) > 1,
1538 self.EINSTANCELAYOUT, instance,
1539 "instance has multiple secondary nodes", code="WARNING")
1541 if not cluster.FillBE(inst_config)[constants.BE_AUTO_BALANCE]:
1542 i_non_a_balanced.append(instance)
1544 for snode in inst_config.secondary_nodes:
1545 _ErrorIf(snode not in node_info and snode not in n_offline,
1546 self.ENODERPC, snode,
1547 "instance %s, connection to secondary node"
1548 " failed", instance)
1550 if snode in node_info:
1551 node_info[snode]['sinst'].append(instance)
1552 if pnode not in node_info[snode]['sinst-by-pnode']:
1553 node_info[snode]['sinst-by-pnode'][pnode] = []
1554 node_info[snode]['sinst-by-pnode'][pnode].append(instance)
1556 if snode in n_offline:
1557 inst_nodes_offline.append(snode)
1559 # warn that the instance lives on offline nodes
1560 _ErrorIf(inst_nodes_offline, self.EINSTANCEBADNODE, instance,
1561 "instance lives on offline node(s) %s",
1562 utils.CommaJoin(inst_nodes_offline))
1564 feedback_fn("* Verifying orphan volumes")
1565 self._VerifyOrphanVolumes(node_vol_should, node_volume)
1567 feedback_fn("* Verifying remaining instances")
1568 self._VerifyOrphanInstances(instancelist, node_instance)
1570 if constants.VERIFY_NPLUSONE_MEM not in self.skip_set:
1571 feedback_fn("* Verifying N+1 Memory redundancy")
1572 self._VerifyNPlusOneMemory(node_info, instance_cfg)
1574 feedback_fn("* Other Notes")
1576 feedback_fn(" - NOTICE: %d non-redundant instance(s) found."
1577 % len(i_non_redundant))
1579 if i_non_a_balanced:
1580 feedback_fn(" - NOTICE: %d non-auto-balanced instance(s) found."
1581 % len(i_non_a_balanced))
1584 feedback_fn(" - NOTICE: %d offline node(s) found." % len(n_offline))
1587 feedback_fn(" - NOTICE: %d drained node(s) found." % len(n_drained))
1591 def HooksCallBack(self, phase, hooks_results, feedback_fn, lu_result):
1592 """Analyze the post-hooks' result
1594 This method analyses the hook result, handles it, and sends some
1595 nicely-formatted feedback back to the user.
1597 @param phase: one of L{constants.HOOKS_PHASE_POST} or
1598 L{constants.HOOKS_PHASE_PRE}; it denotes the hooks phase
1599 @param hooks_results: the results of the multi-node hooks rpc call
1600 @param feedback_fn: function used send feedback back to the caller
1601 @param lu_result: previous Exec result
1602 @return: the new Exec result, based on the previous result
1606 # We only really run POST phase hooks, and are only interested in
1608 if phase == constants.HOOKS_PHASE_POST:
1609 # Used to change hooks' output to proper indentation
1610 indent_re = re.compile('^', re.M)
1611 feedback_fn("* Hooks Results")
1612 assert hooks_results, "invalid result from hooks"
1614 for node_name in hooks_results:
1615 res = hooks_results[node_name]
1617 test = msg and not res.offline
1618 self._ErrorIf(test, self.ENODEHOOKS, node_name,
1619 "Communication failure in hooks execution: %s", msg)
1620 if res.offline or msg:
1621 # No need to investigate payload if node is offline or gave an error.
1622 # override manually lu_result here as _ErrorIf only
1623 # overrides self.bad
1626 for script, hkr, output in res.payload:
1627 test = hkr == constants.HKR_FAIL
1628 self._ErrorIf(test, self.ENODEHOOKS, node_name,
1629 "Script %s failed, output:", script)
1631 output = indent_re.sub(' ', output)
1632 feedback_fn("%s" % output)
1638 class LUVerifyDisks(NoHooksLU):
1639 """Verifies the cluster disks status.
1645 def ExpandNames(self):
1646 self.needed_locks = {
1647 locking.LEVEL_NODE: locking.ALL_SET,
1648 locking.LEVEL_INSTANCE: locking.ALL_SET,
1650 self.share_locks = dict.fromkeys(locking.LEVELS, 1)
1652 def CheckPrereq(self):
1653 """Check prerequisites.
1655 This has no prerequisites.
1660 def Exec(self, feedback_fn):
1661 """Verify integrity of cluster disks.
1663 @rtype: tuple of three items
1664 @return: a tuple of (dict of node-to-node_error, list of instances
1665 which need activate-disks, dict of instance: (node, volume) for
1669 result = res_nodes, res_instances, res_missing = {}, [], {}
1671 vg_name = self.cfg.GetVGName()
1672 nodes = utils.NiceSort(self.cfg.GetNodeList())
1673 instances = [self.cfg.GetInstanceInfo(name)
1674 for name in self.cfg.GetInstanceList()]
1677 for inst in instances:
1679 if (not inst.admin_up or
1680 inst.disk_template not in constants.DTS_NET_MIRROR):
1682 inst.MapLVsByNode(inst_lvs)
1683 # transform { iname: {node: [vol,],},} to {(node, vol): iname}
1684 for node, vol_list in inst_lvs.iteritems():
1685 for vol in vol_list:
1686 nv_dict[(node, vol)] = inst
1691 node_lvs = self.rpc.call_lv_list(nodes, vg_name)
1695 node_res = node_lvs[node]
1696 if node_res.offline:
1698 msg = node_res.fail_msg
1700 logging.warning("Error enumerating LVs on node %s: %s", node, msg)
1701 res_nodes[node] = msg
1704 lvs = node_res.payload
1705 for lv_name, (_, _, lv_online) in lvs.items():
1706 inst = nv_dict.pop((node, lv_name), None)
1707 if (not lv_online and inst is not None
1708 and inst.name not in res_instances):
1709 res_instances.append(inst.name)
1711 # any leftover items in nv_dict are missing LVs, let's arrange the
1713 for key, inst in nv_dict.iteritems():
1714 if inst.name not in res_missing:
1715 res_missing[inst.name] = []
1716 res_missing[inst.name].append(key)
1721 class LURepairDiskSizes(NoHooksLU):
1722 """Verifies the cluster disks sizes.
1725 _OP_REQP = ["instances"]
1728 def ExpandNames(self):
1729 if not isinstance(self.op.instances, list):
1730 raise errors.OpPrereqError("Invalid argument type 'instances'",
1733 if self.op.instances:
1734 self.wanted_names = []
1735 for name in self.op.instances:
1736 full_name = _ExpandInstanceName(self.cfg, name)
1737 self.wanted_names.append(full_name)
1738 self.needed_locks = {
1739 locking.LEVEL_NODE: [],
1740 locking.LEVEL_INSTANCE: self.wanted_names,
1742 self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
1744 self.wanted_names = None
1745 self.needed_locks = {
1746 locking.LEVEL_NODE: locking.ALL_SET,
1747 locking.LEVEL_INSTANCE: locking.ALL_SET,
1749 self.share_locks = dict(((i, 1) for i in locking.LEVELS))
1751 def DeclareLocks(self, level):
1752 if level == locking.LEVEL_NODE and self.wanted_names is not None:
1753 self._LockInstancesNodes(primary_only=True)
1755 def CheckPrereq(self):
1756 """Check prerequisites.
1758 This only checks the optional instance list against the existing names.
1761 if self.wanted_names is None:
1762 self.wanted_names = self.acquired_locks[locking.LEVEL_INSTANCE]
1764 self.wanted_instances = [self.cfg.GetInstanceInfo(name) for name
1765 in self.wanted_names]
1767 def _EnsureChildSizes(self, disk):
1768 """Ensure children of the disk have the needed disk size.
1770 This is valid mainly for DRBD8 and fixes an issue where the
1771 children have smaller disk size.
1773 @param disk: an L{ganeti.objects.Disk} object
1776 if disk.dev_type == constants.LD_DRBD8:
1777 assert disk.children, "Empty children for DRBD8?"
1778 fchild = disk.children[0]
1779 mismatch = fchild.size < disk.size
1781 self.LogInfo("Child disk has size %d, parent %d, fixing",
1782 fchild.size, disk.size)
1783 fchild.size = disk.size
1785 # and we recurse on this child only, not on the metadev
1786 return self._EnsureChildSizes(fchild) or mismatch
1790 def Exec(self, feedback_fn):
1791 """Verify the size of cluster disks.
1794 # TODO: check child disks too
1795 # TODO: check differences in size between primary/secondary nodes
1797 for instance in self.wanted_instances:
1798 pnode = instance.primary_node
1799 if pnode not in per_node_disks:
1800 per_node_disks[pnode] = []
1801 for idx, disk in enumerate(instance.disks):
1802 per_node_disks[pnode].append((instance, idx, disk))
1805 for node, dskl in per_node_disks.items():
1806 newl = [v[2].Copy() for v in dskl]
1808 self.cfg.SetDiskID(dsk, node)
1809 result = self.rpc.call_blockdev_getsizes(node, newl)
1811 self.LogWarning("Failure in blockdev_getsizes call to node"
1812 " %s, ignoring", node)
1814 if len(result.data) != len(dskl):
1815 self.LogWarning("Invalid result from node %s, ignoring node results",
1818 for ((instance, idx, disk), size) in zip(dskl, result.data):
1820 self.LogWarning("Disk %d of instance %s did not return size"
1821 " information, ignoring", idx, instance.name)
1823 if not isinstance(size, (int, long)):
1824 self.LogWarning("Disk %d of instance %s did not return valid"
1825 " size information, ignoring", idx, instance.name)
1828 if size != disk.size:
1829 self.LogInfo("Disk %d of instance %s has mismatched size,"
1830 " correcting: recorded %d, actual %d", idx,
1831 instance.name, disk.size, size)
1833 self.cfg.Update(instance, feedback_fn)
1834 changed.append((instance.name, idx, size))
1835 if self._EnsureChildSizes(disk):
1836 self.cfg.Update(instance, feedback_fn)
1837 changed.append((instance.name, idx, disk.size))
1841 class LURenameCluster(LogicalUnit):
1842 """Rename the cluster.
1845 HPATH = "cluster-rename"
1846 HTYPE = constants.HTYPE_CLUSTER
1849 def BuildHooksEnv(self):
1854 "OP_TARGET": self.cfg.GetClusterName(),
1855 "NEW_NAME": self.op.name,
1857 mn = self.cfg.GetMasterNode()
1858 all_nodes = self.cfg.GetNodeList()
1859 return env, [mn], all_nodes
1861 def CheckPrereq(self):
1862 """Verify that the passed name is a valid one.
1865 hostname = utils.GetHostInfo(self.op.name)
1867 new_name = hostname.name
1868 self.ip = new_ip = hostname.ip
1869 old_name = self.cfg.GetClusterName()
1870 old_ip = self.cfg.GetMasterIP()
1871 if new_name == old_name and new_ip == old_ip:
1872 raise errors.OpPrereqError("Neither the name nor the IP address of the"
1873 " cluster has changed",
1875 if new_ip != old_ip:
1876 if utils.TcpPing(new_ip, constants.DEFAULT_NODED_PORT):
1877 raise errors.OpPrereqError("The given cluster IP address (%s) is"
1878 " reachable on the network. Aborting." %
1879 new_ip, errors.ECODE_NOTUNIQUE)
1881 self.op.name = new_name
1883 def Exec(self, feedback_fn):
1884 """Rename the cluster.
1887 clustername = self.op.name
1890 # shutdown the master IP
1891 master = self.cfg.GetMasterNode()
1892 result = self.rpc.call_node_stop_master(master, False)
1893 result.Raise("Could not disable the master role")
1896 cluster = self.cfg.GetClusterInfo()
1897 cluster.cluster_name = clustername
1898 cluster.master_ip = ip
1899 self.cfg.Update(cluster, feedback_fn)
1901 # update the known hosts file
1902 ssh.WriteKnownHostsFile(self.cfg, constants.SSH_KNOWN_HOSTS_FILE)
1903 node_list = self.cfg.GetNodeList()
1905 node_list.remove(master)
1908 result = self.rpc.call_upload_file(node_list,
1909 constants.SSH_KNOWN_HOSTS_FILE)
1910 for to_node, to_result in result.iteritems():
1911 msg = to_result.fail_msg
1913 msg = ("Copy of file %s to node %s failed: %s" %
1914 (constants.SSH_KNOWN_HOSTS_FILE, to_node, msg))
1915 self.proc.LogWarning(msg)
1918 result = self.rpc.call_node_start_master(master, False, False)
1919 msg = result.fail_msg
1921 self.LogWarning("Could not re-enable the master role on"
1922 " the master, please restart manually: %s", msg)
1925 def _RecursiveCheckIfLVMBased(disk):
1926 """Check if the given disk or its children are lvm-based.
1928 @type disk: L{objects.Disk}
1929 @param disk: the disk to check
1931 @return: boolean indicating whether a LD_LV dev_type was found or not
1935 for chdisk in disk.children:
1936 if _RecursiveCheckIfLVMBased(chdisk):
1938 return disk.dev_type == constants.LD_LV
1941 class LUSetClusterParams(LogicalUnit):
1942 """Change the parameters of the cluster.
1945 HPATH = "cluster-modify"
1946 HTYPE = constants.HTYPE_CLUSTER
1950 def CheckArguments(self):
1954 if not hasattr(self.op, "candidate_pool_size"):
1955 self.op.candidate_pool_size = None
1956 if self.op.candidate_pool_size is not None:
1958 self.op.candidate_pool_size = int(self.op.candidate_pool_size)
1959 except (ValueError, TypeError), err:
1960 raise errors.OpPrereqError("Invalid candidate_pool_size value: %s" %
1961 str(err), errors.ECODE_INVAL)
1962 if self.op.candidate_pool_size < 1:
1963 raise errors.OpPrereqError("At least one master candidate needed",
1966 def ExpandNames(self):
1967 # FIXME: in the future maybe other cluster params won't require checking on
1968 # all nodes to be modified.
1969 self.needed_locks = {
1970 locking.LEVEL_NODE: locking.ALL_SET,
1972 self.share_locks[locking.LEVEL_NODE] = 1
1974 def BuildHooksEnv(self):
1979 "OP_TARGET": self.cfg.GetClusterName(),
1980 "NEW_VG_NAME": self.op.vg_name,
1982 mn = self.cfg.GetMasterNode()
1983 return env, [mn], [mn]
1985 def CheckPrereq(self):
1986 """Check prerequisites.
1988 This checks whether the given params don't conflict and
1989 if the given volume group is valid.
1992 if self.op.vg_name is not None and not self.op.vg_name:
1993 instances = self.cfg.GetAllInstancesInfo().values()
1994 for inst in instances:
1995 for disk in inst.disks:
1996 if _RecursiveCheckIfLVMBased(disk):
1997 raise errors.OpPrereqError("Cannot disable lvm storage while"
1998 " lvm-based instances exist",
2001 node_list = self.acquired_locks[locking.LEVEL_NODE]
2003 # if vg_name not None, checks given volume group on all nodes
2005 vglist = self.rpc.call_vg_list(node_list)
2006 for node in node_list:
2007 msg = vglist[node].fail_msg
2009 # ignoring down node
2010 self.LogWarning("Error while gathering data on node %s"
2011 " (ignoring node): %s", node, msg)
2013 vgstatus = utils.CheckVolumeGroupSize(vglist[node].payload,
2015 constants.MIN_VG_SIZE)
2017 raise errors.OpPrereqError("Error on node '%s': %s" %
2018 (node, vgstatus), errors.ECODE_ENVIRON)
2020 self.cluster = cluster = self.cfg.GetClusterInfo()
2021 # validate params changes
2022 if self.op.beparams:
2023 utils.ForceDictType(self.op.beparams, constants.BES_PARAMETER_TYPES)
2024 self.new_beparams = objects.FillDict(
2025 cluster.beparams[constants.PP_DEFAULT], self.op.beparams)
2027 if self.op.nicparams:
2028 utils.ForceDictType(self.op.nicparams, constants.NICS_PARAMETER_TYPES)
2029 self.new_nicparams = objects.FillDict(
2030 cluster.nicparams[constants.PP_DEFAULT], self.op.nicparams)
2031 objects.NIC.CheckParameterSyntax(self.new_nicparams)
2034 # check all instances for consistency
2035 for instance in self.cfg.GetAllInstancesInfo().values():
2036 for nic_idx, nic in enumerate(instance.nics):
2037 params_copy = copy.deepcopy(nic.nicparams)
2038 params_filled = objects.FillDict(self.new_nicparams, params_copy)
2040 # check parameter syntax
2042 objects.NIC.CheckParameterSyntax(params_filled)
2043 except errors.ConfigurationError, err:
2044 nic_errors.append("Instance %s, nic/%d: %s" %
2045 (instance.name, nic_idx, err))
2047 # if we're moving instances to routed, check that they have an ip
2048 target_mode = params_filled[constants.NIC_MODE]
2049 if target_mode == constants.NIC_MODE_ROUTED and not nic.ip:
2050 nic_errors.append("Instance %s, nic/%d: routed nick with no ip" %
2051 (instance.name, nic_idx))
2053 raise errors.OpPrereqError("Cannot apply the change, errors:\n%s" %
2054 "\n".join(nic_errors))
2056 # hypervisor list/parameters
2057 self.new_hvparams = objects.FillDict(cluster.hvparams, {})
2058 if self.op.hvparams:
2059 if not isinstance(self.op.hvparams, dict):
2060 raise errors.OpPrereqError("Invalid 'hvparams' parameter on input",
2062 for hv_name, hv_dict in self.op.hvparams.items():
2063 if hv_name not in self.new_hvparams:
2064 self.new_hvparams[hv_name] = hv_dict
2066 self.new_hvparams[hv_name].update(hv_dict)
2068 # os hypervisor parameters
2069 self.new_os_hvp = objects.FillDict(cluster.os_hvp, {})
2071 if not isinstance(self.op.os_hvp, dict):
2072 raise errors.OpPrereqError("Invalid 'os_hvp' parameter on input",
2074 for os_name, hvs in self.op.os_hvp.items():
2075 if not isinstance(hvs, dict):
2076 raise errors.OpPrereqError(("Invalid 'os_hvp' parameter on"
2077 " input"), errors.ECODE_INVAL)
2078 if os_name not in self.new_os_hvp:
2079 self.new_os_hvp[os_name] = hvs
2081 for hv_name, hv_dict in hvs.items():
2082 if hv_name not in self.new_os_hvp[os_name]:
2083 self.new_os_hvp[os_name][hv_name] = hv_dict
2085 self.new_os_hvp[os_name][hv_name].update(hv_dict)
2087 if self.op.enabled_hypervisors is not None:
2088 self.hv_list = self.op.enabled_hypervisors
2089 if not self.hv_list:
2090 raise errors.OpPrereqError("Enabled hypervisors list must contain at"
2091 " least one member",
2093 invalid_hvs = set(self.hv_list) - constants.HYPER_TYPES
2095 raise errors.OpPrereqError("Enabled hypervisors contains invalid"
2097 utils.CommaJoin(invalid_hvs),
2100 self.hv_list = cluster.enabled_hypervisors
2102 if self.op.hvparams or self.op.enabled_hypervisors is not None:
2103 # either the enabled list has changed, or the parameters have, validate
2104 for hv_name, hv_params in self.new_hvparams.items():
2105 if ((self.op.hvparams and hv_name in self.op.hvparams) or
2106 (self.op.enabled_hypervisors and
2107 hv_name in self.op.enabled_hypervisors)):
2108 # either this is a new hypervisor, or its parameters have changed
2109 hv_class = hypervisor.GetHypervisor(hv_name)
2110 utils.ForceDictType(hv_params, constants.HVS_PARAMETER_TYPES)
2111 hv_class.CheckParameterSyntax(hv_params)
2112 _CheckHVParams(self, node_list, hv_name, hv_params)
2115 # no need to check any newly-enabled hypervisors, since the
2116 # defaults have already been checked in the above code-block
2117 for os_name, os_hvp in self.new_os_hvp.items():
2118 for hv_name, hv_params in os_hvp.items():
2119 utils.ForceDictType(hv_params, constants.HVS_PARAMETER_TYPES)
2120 # we need to fill in the new os_hvp on top of the actual hv_p
2121 cluster_defaults = self.new_hvparams.get(hv_name, {})
2122 new_osp = objects.FillDict(cluster_defaults, hv_params)
2123 hv_class = hypervisor.GetHypervisor(hv_name)
2124 hv_class.CheckParameterSyntax(new_osp)
2125 _CheckHVParams(self, node_list, hv_name, new_osp)
2128 def Exec(self, feedback_fn):
2129 """Change the parameters of the cluster.
2132 if self.op.vg_name is not None:
2133 new_volume = self.op.vg_name
2136 if new_volume != self.cfg.GetVGName():
2137 self.cfg.SetVGName(new_volume)
2139 feedback_fn("Cluster LVM configuration already in desired"
2140 " state, not changing")
2141 if self.op.hvparams:
2142 self.cluster.hvparams = self.new_hvparams
2144 self.cluster.os_hvp = self.new_os_hvp
2145 if self.op.enabled_hypervisors is not None:
2146 self.cluster.enabled_hypervisors = self.op.enabled_hypervisors
2147 if self.op.beparams:
2148 self.cluster.beparams[constants.PP_DEFAULT] = self.new_beparams
2149 if self.op.nicparams:
2150 self.cluster.nicparams[constants.PP_DEFAULT] = self.new_nicparams
2152 if self.op.candidate_pool_size is not None:
2153 self.cluster.candidate_pool_size = self.op.candidate_pool_size
2154 # we need to update the pool size here, otherwise the save will fail
2155 _AdjustCandidatePool(self, [])
2157 self.cfg.Update(self.cluster, feedback_fn)
2160 def _RedistributeAncillaryFiles(lu, additional_nodes=None):
2161 """Distribute additional files which are part of the cluster configuration.
2163 ConfigWriter takes care of distributing the config and ssconf files, but
2164 there are more files which should be distributed to all nodes. This function
2165 makes sure those are copied.
2167 @param lu: calling logical unit
2168 @param additional_nodes: list of nodes not in the config to distribute to
2171 # 1. Gather target nodes
2172 myself = lu.cfg.GetNodeInfo(lu.cfg.GetMasterNode())
2173 dist_nodes = lu.cfg.GetOnlineNodeList()
2174 if additional_nodes is not None:
2175 dist_nodes.extend(additional_nodes)
2176 if myself.name in dist_nodes:
2177 dist_nodes.remove(myself.name)
2179 # 2. Gather files to distribute
2180 dist_files = set([constants.ETC_HOSTS,
2181 constants.SSH_KNOWN_HOSTS_FILE,
2182 constants.RAPI_CERT_FILE,
2183 constants.RAPI_USERS_FILE,
2184 constants.HMAC_CLUSTER_KEY,
2187 enabled_hypervisors = lu.cfg.GetClusterInfo().enabled_hypervisors
2188 for hv_name in enabled_hypervisors:
2189 hv_class = hypervisor.GetHypervisor(hv_name)
2190 dist_files.update(hv_class.GetAncillaryFiles())
2192 # 3. Perform the files upload
2193 for fname in dist_files:
2194 if os.path.exists(fname):
2195 result = lu.rpc.call_upload_file(dist_nodes, fname)
2196 for to_node, to_result in result.items():
2197 msg = to_result.fail_msg
2199 msg = ("Copy of file %s to node %s failed: %s" %
2200 (fname, to_node, msg))
2201 lu.proc.LogWarning(msg)
2204 class LURedistributeConfig(NoHooksLU):
2205 """Force the redistribution of cluster configuration.
2207 This is a very simple LU.
2213 def ExpandNames(self):
2214 self.needed_locks = {
2215 locking.LEVEL_NODE: locking.ALL_SET,
2217 self.share_locks[locking.LEVEL_NODE] = 1
2219 def CheckPrereq(self):
2220 """Check prerequisites.
2224 def Exec(self, feedback_fn):
2225 """Redistribute the configuration.
2228 self.cfg.Update(self.cfg.GetClusterInfo(), feedback_fn)
2229 _RedistributeAncillaryFiles(self)
2232 def _WaitForSync(lu, instance, oneshot=False):
2233 """Sleep and poll for an instance's disk to sync.
2236 if not instance.disks:
2240 lu.proc.LogInfo("Waiting for instance %s to sync disks." % instance.name)
2242 node = instance.primary_node
2244 for dev in instance.disks:
2245 lu.cfg.SetDiskID(dev, node)
2247 # TODO: Convert to utils.Retry
2250 degr_retries = 10 # in seconds, as we sleep 1 second each time
2254 cumul_degraded = False
2255 rstats = lu.rpc.call_blockdev_getmirrorstatus(node, instance.disks)
2256 msg = rstats.fail_msg
2258 lu.LogWarning("Can't get any data from node %s: %s", node, msg)
2261 raise errors.RemoteError("Can't contact node %s for mirror data,"
2262 " aborting." % node)
2265 rstats = rstats.payload
2267 for i, mstat in enumerate(rstats):
2269 lu.LogWarning("Can't compute data for node %s/%s",
2270 node, instance.disks[i].iv_name)
2273 cumul_degraded = (cumul_degraded or
2274 (mstat.is_degraded and mstat.sync_percent is None))
2275 if mstat.sync_percent is not None:
2277 if mstat.estimated_time is not None:
2278 rem_time = "%d estimated seconds remaining" % mstat.estimated_time
2279 max_time = mstat.estimated_time
2281 rem_time = "no time estimate"
2282 lu.proc.LogInfo("- device %s: %5.2f%% done, %s" %
2283 (instance.disks[i].iv_name, mstat.sync_percent,
2286 # if we're done but degraded, let's do a few small retries, to
2287 # make sure we see a stable and not transient situation; therefore
2288 # we force restart of the loop
2289 if (done or oneshot) and cumul_degraded and degr_retries > 0:
2290 logging.info("Degraded disks found, %d retries left", degr_retries)
2298 time.sleep(min(60, max_time))
2301 lu.proc.LogInfo("Instance %s's disks are in sync." % instance.name)
2302 return not cumul_degraded
2305 def _CheckDiskConsistency(lu, dev, node, on_primary, ldisk=False):
2306 """Check that mirrors are not degraded.
2308 The ldisk parameter, if True, will change the test from the
2309 is_degraded attribute (which represents overall non-ok status for
2310 the device(s)) to the ldisk (representing the local storage status).
2313 lu.cfg.SetDiskID(dev, node)
2317 if on_primary or dev.AssembleOnSecondary():
2318 rstats = lu.rpc.call_blockdev_find(node, dev)
2319 msg = rstats.fail_msg
2321 lu.LogWarning("Can't find disk on node %s: %s", node, msg)
2323 elif not rstats.payload:
2324 lu.LogWarning("Can't find disk on node %s", node)
2328 result = result and rstats.payload.ldisk_status == constants.LDS_OKAY
2330 result = result and not rstats.payload.is_degraded
2333 for child in dev.children:
2334 result = result and _CheckDiskConsistency(lu, child, node, on_primary)
2339 class LUDiagnoseOS(NoHooksLU):
2340 """Logical unit for OS diagnose/query.
2343 _OP_REQP = ["output_fields", "names"]
2345 _FIELDS_STATIC = utils.FieldSet()
2346 _FIELDS_DYNAMIC = utils.FieldSet("name", "valid", "node_status", "variants")
2347 # Fields that need calculation of global os validity
2348 _FIELDS_NEEDVALID = frozenset(["valid", "variants"])
2350 def ExpandNames(self):
2352 raise errors.OpPrereqError("Selective OS query not supported",
2355 _CheckOutputFields(static=self._FIELDS_STATIC,
2356 dynamic=self._FIELDS_DYNAMIC,
2357 selected=self.op.output_fields)
2359 # Lock all nodes, in shared mode
2360 # Temporary removal of locks, should be reverted later
2361 # TODO: reintroduce locks when they are lighter-weight
2362 self.needed_locks = {}
2363 #self.share_locks[locking.LEVEL_NODE] = 1
2364 #self.needed_locks[locking.LEVEL_NODE] = locking.ALL_SET
2366 def CheckPrereq(self):
2367 """Check prerequisites.
2372 def _DiagnoseByOS(rlist):
2373 """Remaps a per-node return list into an a per-os per-node dictionary
2375 @param rlist: a map with node names as keys and OS objects as values
2378 @return: a dictionary with osnames as keys and as value another map, with
2379 nodes as keys and tuples of (path, status, diagnose) as values, eg::
2381 {"debian-etch": {"node1": [(/usr/lib/..., True, ""),
2382 (/srv/..., False, "invalid api")],
2383 "node2": [(/srv/..., True, "")]}
2388 # we build here the list of nodes that didn't fail the RPC (at RPC
2389 # level), so that nodes with a non-responding node daemon don't
2390 # make all OSes invalid
2391 good_nodes = [node_name for node_name in rlist
2392 if not rlist[node_name].fail_msg]
2393 for node_name, nr in rlist.items():
2394 if nr.fail_msg or not nr.payload:
2396 for name, path, status, diagnose, variants in nr.payload:
2397 if name not in all_os:
2398 # build a list of nodes for this os containing empty lists
2399 # for each node in node_list
2401 for nname in good_nodes:
2402 all_os[name][nname] = []
2403 all_os[name][node_name].append((path, status, diagnose, variants))
2406 def Exec(self, feedback_fn):
2407 """Compute the list of OSes.
2410 valid_nodes = [node for node in self.cfg.GetOnlineNodeList()]
2411 node_data = self.rpc.call_os_diagnose(valid_nodes)
2412 pol = self._DiagnoseByOS(node_data)
2414 calc_valid = self._FIELDS_NEEDVALID.intersection(self.op.output_fields)
2415 calc_variants = "variants" in self.op.output_fields
2417 for os_name, os_data in pol.items():
2422 for osl in os_data.values():
2423 valid = valid and osl and osl[0][1]
2428 node_variants = osl[0][3]
2429 if variants is None:
2430 variants = node_variants
2432 variants = [v for v in variants if v in node_variants]
2434 for field in self.op.output_fields:
2437 elif field == "valid":
2439 elif field == "node_status":
2440 # this is just a copy of the dict
2442 for node_name, nos_list in os_data.items():
2443 val[node_name] = nos_list
2444 elif field == "variants":
2447 raise errors.ParameterError(field)
2454 class LURemoveNode(LogicalUnit):
2455 """Logical unit for removing a node.
2458 HPATH = "node-remove"
2459 HTYPE = constants.HTYPE_NODE
2460 _OP_REQP = ["node_name"]
2462 def BuildHooksEnv(self):
2465 This doesn't run on the target node in the pre phase as a failed
2466 node would then be impossible to remove.
2470 "OP_TARGET": self.op.node_name,
2471 "NODE_NAME": self.op.node_name,
2473 all_nodes = self.cfg.GetNodeList()
2475 all_nodes.remove(self.op.node_name)
2477 logging.warning("Node %s which is about to be removed not found"
2478 " in the all nodes list", self.op.node_name)
2479 return env, all_nodes, all_nodes
2481 def CheckPrereq(self):
2482 """Check prerequisites.
2485 - the node exists in the configuration
2486 - it does not have primary or secondary instances
2487 - it's not the master
2489 Any errors are signaled by raising errors.OpPrereqError.
2492 self.op.node_name = _ExpandNodeName(self.cfg, self.op.node_name)
2493 node = self.cfg.GetNodeInfo(self.op.node_name)
2494 assert node is not None
2496 instance_list = self.cfg.GetInstanceList()
2498 masternode = self.cfg.GetMasterNode()
2499 if node.name == masternode:
2500 raise errors.OpPrereqError("Node is the master node,"
2501 " you need to failover first.",
2504 for instance_name in instance_list:
2505 instance = self.cfg.GetInstanceInfo(instance_name)
2506 if node.name in instance.all_nodes:
2507 raise errors.OpPrereqError("Instance %s is still running on the node,"
2508 " please remove first." % instance_name,
2510 self.op.node_name = node.name
2513 def Exec(self, feedback_fn):
2514 """Removes the node from the cluster.
2518 logging.info("Stopping the node daemon and removing configs from node %s",
2521 modify_ssh_setup = self.cfg.GetClusterInfo().modify_ssh_setup
2523 # Promote nodes to master candidate as needed
2524 _AdjustCandidatePool(self, exceptions=[node.name])
2525 self.context.RemoveNode(node.name)
2527 # Run post hooks on the node before it's removed
2528 hm = self.proc.hmclass(self.rpc.call_hooks_runner, self)
2530 hm.RunPhase(constants.HOOKS_PHASE_POST, [node.name])
2532 # pylint: disable-msg=W0702
2533 self.LogWarning("Errors occurred running hooks on %s" % node.name)
2535 result = self.rpc.call_node_leave_cluster(node.name, modify_ssh_setup)
2536 msg = result.fail_msg
2538 self.LogWarning("Errors encountered on the remote node while leaving"
2539 " the cluster: %s", msg)
2542 class LUQueryNodes(NoHooksLU):
2543 """Logical unit for querying nodes.
2546 # pylint: disable-msg=W0142
2547 _OP_REQP = ["output_fields", "names", "use_locking"]
2550 _SIMPLE_FIELDS = ["name", "serial_no", "ctime", "mtime", "uuid",
2551 "master_candidate", "offline", "drained"]
2553 _FIELDS_DYNAMIC = utils.FieldSet(
2555 "mtotal", "mnode", "mfree",
2557 "ctotal", "cnodes", "csockets",
2560 _FIELDS_STATIC = utils.FieldSet(*[
2561 "pinst_cnt", "sinst_cnt",
2562 "pinst_list", "sinst_list",
2563 "pip", "sip", "tags",
2565 "role"] + _SIMPLE_FIELDS
2568 def ExpandNames(self):
2569 _CheckOutputFields(static=self._FIELDS_STATIC,
2570 dynamic=self._FIELDS_DYNAMIC,
2571 selected=self.op.output_fields)
2573 self.needed_locks = {}
2574 self.share_locks[locking.LEVEL_NODE] = 1
2577 self.wanted = _GetWantedNodes(self, self.op.names)
2579 self.wanted = locking.ALL_SET
2581 self.do_node_query = self._FIELDS_STATIC.NonMatching(self.op.output_fields)
2582 self.do_locking = self.do_node_query and self.op.use_locking
2584 # if we don't request only static fields, we need to lock the nodes
2585 self.needed_locks[locking.LEVEL_NODE] = self.wanted
2587 def CheckPrereq(self):
2588 """Check prerequisites.
2591 # The validation of the node list is done in the _GetWantedNodes,
2592 # if non empty, and if empty, there's no validation to do
2595 def Exec(self, feedback_fn):
2596 """Computes the list of nodes and their attributes.
2599 all_info = self.cfg.GetAllNodesInfo()
2601 nodenames = self.acquired_locks[locking.LEVEL_NODE]
2602 elif self.wanted != locking.ALL_SET:
2603 nodenames = self.wanted
2604 missing = set(nodenames).difference(all_info.keys())
2606 raise errors.OpExecError(
2607 "Some nodes were removed before retrieving their data: %s" % missing)
2609 nodenames = all_info.keys()
2611 nodenames = utils.NiceSort(nodenames)
2612 nodelist = [all_info[name] for name in nodenames]
2614 # begin data gathering
2616 if self.do_node_query:
2618 node_data = self.rpc.call_node_info(nodenames, self.cfg.GetVGName(),
2619 self.cfg.GetHypervisorType())
2620 for name in nodenames:
2621 nodeinfo = node_data[name]
2622 if not nodeinfo.fail_msg and nodeinfo.payload:
2623 nodeinfo = nodeinfo.payload
2624 fn = utils.TryConvert
2626 "mtotal": fn(int, nodeinfo.get('memory_total', None)),
2627 "mnode": fn(int, nodeinfo.get('memory_dom0', None)),
2628 "mfree": fn(int, nodeinfo.get('memory_free', None)),
2629 "dtotal": fn(int, nodeinfo.get('vg_size', None)),
2630 "dfree": fn(int, nodeinfo.get('vg_free', None)),
2631 "ctotal": fn(int, nodeinfo.get('cpu_total', None)),
2632 "bootid": nodeinfo.get('bootid', None),
2633 "cnodes": fn(int, nodeinfo.get('cpu_nodes', None)),
2634 "csockets": fn(int, nodeinfo.get('cpu_sockets', None)),
2637 live_data[name] = {}
2639 live_data = dict.fromkeys(nodenames, {})
2641 node_to_primary = dict([(name, set()) for name in nodenames])
2642 node_to_secondary = dict([(name, set()) for name in nodenames])
2644 inst_fields = frozenset(("pinst_cnt", "pinst_list",
2645 "sinst_cnt", "sinst_list"))
2646 if inst_fields & frozenset(self.op.output_fields):
2647 inst_data = self.cfg.GetAllInstancesInfo()
2649 for inst in inst_data.values():
2650 if inst.primary_node in node_to_primary:
2651 node_to_primary[inst.primary_node].add(inst.name)
2652 for secnode in inst.secondary_nodes:
2653 if secnode in node_to_secondary:
2654 node_to_secondary[secnode].add(inst.name)
2656 master_node = self.cfg.GetMasterNode()
2658 # end data gathering
2661 for node in nodelist:
2663 for field in self.op.output_fields:
2664 if field in self._SIMPLE_FIELDS:
2665 val = getattr(node, field)
2666 elif field == "pinst_list":
2667 val = list(node_to_primary[node.name])
2668 elif field == "sinst_list":
2669 val = list(node_to_secondary[node.name])
2670 elif field == "pinst_cnt":
2671 val = len(node_to_primary[node.name])
2672 elif field == "sinst_cnt":
2673 val = len(node_to_secondary[node.name])
2674 elif field == "pip":
2675 val = node.primary_ip
2676 elif field == "sip":
2677 val = node.secondary_ip
2678 elif field == "tags":
2679 val = list(node.GetTags())
2680 elif field == "master":
2681 val = node.name == master_node
2682 elif self._FIELDS_DYNAMIC.Matches(field):
2683 val = live_data[node.name].get(field, None)
2684 elif field == "role":
2685 if node.name == master_node:
2687 elif node.master_candidate:
2696 raise errors.ParameterError(field)
2697 node_output.append(val)
2698 output.append(node_output)
2703 class LUQueryNodeVolumes(NoHooksLU):
2704 """Logical unit for getting volumes on node(s).
2707 _OP_REQP = ["nodes", "output_fields"]
2709 _FIELDS_DYNAMIC = utils.FieldSet("phys", "vg", "name", "size", "instance")
2710 _FIELDS_STATIC = utils.FieldSet("node")
2712 def ExpandNames(self):
2713 _CheckOutputFields(static=self._FIELDS_STATIC,
2714 dynamic=self._FIELDS_DYNAMIC,
2715 selected=self.op.output_fields)
2717 self.needed_locks = {}
2718 self.share_locks[locking.LEVEL_NODE] = 1
2719 if not self.op.nodes:
2720 self.needed_locks[locking.LEVEL_NODE] = locking.ALL_SET
2722 self.needed_locks[locking.LEVEL_NODE] = \
2723 _GetWantedNodes(self, self.op.nodes)
2725 def CheckPrereq(self):
2726 """Check prerequisites.
2728 This checks that the fields required are valid output fields.
2731 self.nodes = self.acquired_locks[locking.LEVEL_NODE]
2733 def Exec(self, feedback_fn):
2734 """Computes the list of nodes and their attributes.
2737 nodenames = self.nodes
2738 volumes = self.rpc.call_node_volumes(nodenames)
2740 ilist = [self.cfg.GetInstanceInfo(iname) for iname
2741 in self.cfg.GetInstanceList()]
2743 lv_by_node = dict([(inst, inst.MapLVsByNode()) for inst in ilist])
2746 for node in nodenames:
2747 nresult = volumes[node]
2750 msg = nresult.fail_msg
2752 self.LogWarning("Can't compute volume data on node %s: %s", node, msg)
2755 node_vols = nresult.payload[:]
2756 node_vols.sort(key=lambda vol: vol['dev'])
2758 for vol in node_vols:
2760 for field in self.op.output_fields:
2763 elif field == "phys":
2767 elif field == "name":
2769 elif field == "size":
2770 val = int(float(vol['size']))
2771 elif field == "instance":
2773 if node not in lv_by_node[inst]:
2775 if vol['name'] in lv_by_node[inst][node]:
2781 raise errors.ParameterError(field)
2782 node_output.append(str(val))
2784 output.append(node_output)
2789 class LUQueryNodeStorage(NoHooksLU):
2790 """Logical unit for getting information on storage units on node(s).
2793 _OP_REQP = ["nodes", "storage_type", "output_fields"]
2795 _FIELDS_STATIC = utils.FieldSet(constants.SF_NODE)
2797 def ExpandNames(self):
2798 storage_type = self.op.storage_type
2800 if storage_type not in constants.VALID_STORAGE_TYPES:
2801 raise errors.OpPrereqError("Unknown storage type: %s" % storage_type,
2804 _CheckOutputFields(static=self._FIELDS_STATIC,
2805 dynamic=utils.FieldSet(*constants.VALID_STORAGE_FIELDS),
2806 selected=self.op.output_fields)
2808 self.needed_locks = {}
2809 self.share_locks[locking.LEVEL_NODE] = 1
2812 self.needed_locks[locking.LEVEL_NODE] = \
2813 _GetWantedNodes(self, self.op.nodes)
2815 self.needed_locks[locking.LEVEL_NODE] = locking.ALL_SET
2817 def CheckPrereq(self):
2818 """Check prerequisites.
2820 This checks that the fields required are valid output fields.
2823 self.op.name = getattr(self.op, "name", None)
2825 self.nodes = self.acquired_locks[locking.LEVEL_NODE]
2827 def Exec(self, feedback_fn):
2828 """Computes the list of nodes and their attributes.
2831 # Always get name to sort by
2832 if constants.SF_NAME in self.op.output_fields:
2833 fields = self.op.output_fields[:]
2835 fields = [constants.SF_NAME] + self.op.output_fields
2837 # Never ask for node or type as it's only known to the LU
2838 for extra in [constants.SF_NODE, constants.SF_TYPE]:
2839 while extra in fields:
2840 fields.remove(extra)
2842 field_idx = dict([(name, idx) for (idx, name) in enumerate(fields)])
2843 name_idx = field_idx[constants.SF_NAME]
2845 st_args = _GetStorageTypeArgs(self.cfg, self.op.storage_type)
2846 data = self.rpc.call_storage_list(self.nodes,
2847 self.op.storage_type, st_args,
2848 self.op.name, fields)
2852 for node in utils.NiceSort(self.nodes):
2853 nresult = data[node]
2857 msg = nresult.fail_msg
2859 self.LogWarning("Can't get storage data from node %s: %s", node, msg)
2862 rows = dict([(row[name_idx], row) for row in nresult.payload])
2864 for name in utils.NiceSort(rows.keys()):
2869 for field in self.op.output_fields:
2870 if field == constants.SF_NODE:
2872 elif field == constants.SF_TYPE:
2873 val = self.op.storage_type
2874 elif field in field_idx:
2875 val = row[field_idx[field]]
2877 raise errors.ParameterError(field)
2886 class LUModifyNodeStorage(NoHooksLU):
2887 """Logical unit for modifying a storage volume on a node.
2890 _OP_REQP = ["node_name", "storage_type", "name", "changes"]
2893 def CheckArguments(self):
2894 self.opnode_name = _ExpandNodeName(self.cfg, self.op.node_name)
2896 storage_type = self.op.storage_type
2897 if storage_type not in constants.VALID_STORAGE_TYPES:
2898 raise errors.OpPrereqError("Unknown storage type: %s" % storage_type,
2901 def ExpandNames(self):
2902 self.needed_locks = {
2903 locking.LEVEL_NODE: self.op.node_name,
2906 def CheckPrereq(self):
2907 """Check prerequisites.
2910 storage_type = self.op.storage_type
2913 modifiable = constants.MODIFIABLE_STORAGE_FIELDS[storage_type]
2915 raise errors.OpPrereqError("Storage units of type '%s' can not be"
2916 " modified" % storage_type,
2919 diff = set(self.op.changes.keys()) - modifiable
2921 raise errors.OpPrereqError("The following fields can not be modified for"
2922 " storage units of type '%s': %r" %
2923 (storage_type, list(diff)),
2926 def Exec(self, feedback_fn):
2927 """Computes the list of nodes and their attributes.
2930 st_args = _GetStorageTypeArgs(self.cfg, self.op.storage_type)
2931 result = self.rpc.call_storage_modify(self.op.node_name,
2932 self.op.storage_type, st_args,
2933 self.op.name, self.op.changes)
2934 result.Raise("Failed to modify storage unit '%s' on %s" %
2935 (self.op.name, self.op.node_name))
2938 class LUAddNode(LogicalUnit):
2939 """Logical unit for adding node to the cluster.
2943 HTYPE = constants.HTYPE_NODE
2944 _OP_REQP = ["node_name"]
2946 def CheckArguments(self):
2947 # validate/normalize the node name
2948 self.op.node_name = utils.HostInfo.NormalizeName(self.op.node_name)
2950 def BuildHooksEnv(self):
2953 This will run on all nodes before, and on all nodes + the new node after.
2957 "OP_TARGET": self.op.node_name,
2958 "NODE_NAME": self.op.node_name,
2959 "NODE_PIP": self.op.primary_ip,
2960 "NODE_SIP": self.op.secondary_ip,
2962 nodes_0 = self.cfg.GetNodeList()
2963 nodes_1 = nodes_0 + [self.op.node_name, ]
2964 return env, nodes_0, nodes_1
2966 def CheckPrereq(self):
2967 """Check prerequisites.
2970 - the new node is not already in the config
2972 - its parameters (single/dual homed) matches the cluster
2974 Any errors are signaled by raising errors.OpPrereqError.
2977 node_name = self.op.node_name
2980 dns_data = utils.GetHostInfo(node_name)
2982 node = dns_data.name
2983 primary_ip = self.op.primary_ip = dns_data.ip
2984 secondary_ip = getattr(self.op, "secondary_ip", None)
2985 if secondary_ip is None:
2986 secondary_ip = primary_ip
2987 if not utils.IsValidIP(secondary_ip):
2988 raise errors.OpPrereqError("Invalid secondary IP given",
2990 self.op.secondary_ip = secondary_ip
2992 node_list = cfg.GetNodeList()
2993 if not self.op.readd and node in node_list:
2994 raise errors.OpPrereqError("Node %s is already in the configuration" %
2995 node, errors.ECODE_EXISTS)
2996 elif self.op.readd and node not in node_list:
2997 raise errors.OpPrereqError("Node %s is not in the configuration" % node,
3000 for existing_node_name in node_list:
3001 existing_node = cfg.GetNodeInfo(existing_node_name)
3003 if self.op.readd and node == existing_node_name:
3004 if (existing_node.primary_ip != primary_ip or
3005 existing_node.secondary_ip != secondary_ip):
3006 raise errors.OpPrereqError("Readded node doesn't have the same IP"
3007 " address configuration as before",
3011 if (existing_node.primary_ip == primary_ip or
3012 existing_node.secondary_ip == primary_ip or
3013 existing_node.primary_ip == secondary_ip or
3014 existing_node.secondary_ip == secondary_ip):
3015 raise errors.OpPrereqError("New node ip address(es) conflict with"
3016 " existing node %s" % existing_node.name,
3017 errors.ECODE_NOTUNIQUE)
3019 # check that the type of the node (single versus dual homed) is the
3020 # same as for the master
3021 myself = cfg.GetNodeInfo(self.cfg.GetMasterNode())
3022 master_singlehomed = myself.secondary_ip == myself.primary_ip
3023 newbie_singlehomed = secondary_ip == primary_ip
3024 if master_singlehomed != newbie_singlehomed:
3025 if master_singlehomed:
3026 raise errors.OpPrereqError("The master has no private ip but the"
3027 " new node has one",
3030 raise errors.OpPrereqError("The master has a private ip but the"
3031 " new node doesn't have one",
3034 # checks reachability
3035 if not utils.TcpPing(primary_ip, constants.DEFAULT_NODED_PORT):
3036 raise errors.OpPrereqError("Node not reachable by ping",
3037 errors.ECODE_ENVIRON)
3039 if not newbie_singlehomed:
3040 # check reachability from my secondary ip to newbie's secondary ip
3041 if not utils.TcpPing(secondary_ip, constants.DEFAULT_NODED_PORT,
3042 source=myself.secondary_ip):
3043 raise errors.OpPrereqError("Node secondary ip not reachable by TCP"
3044 " based ping to noded port",
3045 errors.ECODE_ENVIRON)
3052 self.master_candidate = _DecideSelfPromotion(self, exceptions=exceptions)
3055 self.new_node = self.cfg.GetNodeInfo(node)
3056 assert self.new_node is not None, "Can't retrieve locked node %s" % node
3058 self.new_node = objects.Node(name=node,
3059 primary_ip=primary_ip,
3060 secondary_ip=secondary_ip,
3061 master_candidate=self.master_candidate,
3062 offline=False, drained=False)
3064 def Exec(self, feedback_fn):
3065 """Adds the new node to the cluster.
3068 new_node = self.new_node
3069 node = new_node.name
3071 # for re-adds, reset the offline/drained/master-candidate flags;
3072 # we need to reset here, otherwise offline would prevent RPC calls
3073 # later in the procedure; this also means that if the re-add
3074 # fails, we are left with a non-offlined, broken node
3076 new_node.drained = new_node.offline = False # pylint: disable-msg=W0201
3077 self.LogInfo("Readding a node, the offline/drained flags were reset")
3078 # if we demote the node, we do cleanup later in the procedure
3079 new_node.master_candidate = self.master_candidate
3081 # notify the user about any possible mc promotion
3082 if new_node.master_candidate:
3083 self.LogInfo("Node will be a master candidate")
3085 # check connectivity
3086 result = self.rpc.call_version([node])[node]
3087 result.Raise("Can't get version information from node %s" % node)
3088 if constants.PROTOCOL_VERSION == result.payload:
3089 logging.info("Communication to node %s fine, sw version %s match",
3090 node, result.payload)
3092 raise errors.OpExecError("Version mismatch master version %s,"
3093 " node version %s" %
3094 (constants.PROTOCOL_VERSION, result.payload))
3097 if self.cfg.GetClusterInfo().modify_ssh_setup:
3098 logging.info("Copy ssh key to node %s", node)
3099 priv_key, pub_key, _ = ssh.GetUserFiles(constants.GANETI_RUNAS)
3101 keyfiles = [constants.SSH_HOST_DSA_PRIV, constants.SSH_HOST_DSA_PUB,
3102 constants.SSH_HOST_RSA_PRIV, constants.SSH_HOST_RSA_PUB,
3106 keyarray.append(utils.ReadFile(i))
3108 result = self.rpc.call_node_add(node, keyarray[0], keyarray[1],
3109 keyarray[2], keyarray[3], keyarray[4],
3111 result.Raise("Cannot transfer ssh keys to the new node")
3113 # Add node to our /etc/hosts, and add key to known_hosts
3114 if self.cfg.GetClusterInfo().modify_etc_hosts:
3115 utils.AddHostToEtcHosts(new_node.name)
3117 if new_node.secondary_ip != new_node.primary_ip:
3118 result = self.rpc.call_node_has_ip_address(new_node.name,
3119 new_node.secondary_ip)
3120 result.Raise("Failure checking secondary ip on node %s" % new_node.name,
3121 prereq=True, ecode=errors.ECODE_ENVIRON)
3122 if not result.payload:
3123 raise errors.OpExecError("Node claims it doesn't have the secondary ip"
3124 " you gave (%s). Please fix and re-run this"
3125 " command." % new_node.secondary_ip)
3127 node_verify_list = [self.cfg.GetMasterNode()]
3128 node_verify_param = {
3129 constants.NV_NODELIST: [node],
3130 # TODO: do a node-net-test as well?
3133 result = self.rpc.call_node_verify(node_verify_list, node_verify_param,
3134 self.cfg.GetClusterName())
3135 for verifier in node_verify_list:
3136 result[verifier].Raise("Cannot communicate with node %s" % verifier)
3137 nl_payload = result[verifier].payload[constants.NV_NODELIST]
3139 for failed in nl_payload:
3140 feedback_fn("ssh/hostname verification failed"
3141 " (checking from %s): %s" %
3142 (verifier, nl_payload[failed]))
3143 raise errors.OpExecError("ssh/hostname verification failed.")
3146 _RedistributeAncillaryFiles(self)
3147 self.context.ReaddNode(new_node)
3148 # make sure we redistribute the config
3149 self.cfg.Update(new_node, feedback_fn)
3150 # and make sure the new node will not have old files around
3151 if not new_node.master_candidate:
3152 result = self.rpc.call_node_demote_from_mc(new_node.name)
3153 msg = result.fail_msg
3155 self.LogWarning("Node failed to demote itself from master"
3156 " candidate status: %s" % msg)
3158 _RedistributeAncillaryFiles(self, additional_nodes=[node])
3159 self.context.AddNode(new_node, self.proc.GetECId())
3162 class LUSetNodeParams(LogicalUnit):
3163 """Modifies the parameters of a node.
3166 HPATH = "node-modify"
3167 HTYPE = constants.HTYPE_NODE
3168 _OP_REQP = ["node_name"]
3171 def CheckArguments(self):
3172 self.op.node_name = _ExpandNodeName(self.cfg, self.op.node_name)
3173 _CheckBooleanOpField(self.op, 'master_candidate')
3174 _CheckBooleanOpField(self.op, 'offline')
3175 _CheckBooleanOpField(self.op, 'drained')
3176 _CheckBooleanOpField(self.op, 'auto_promote')
3177 all_mods = [self.op.offline, self.op.master_candidate, self.op.drained]
3178 if all_mods.count(None) == 3:
3179 raise errors.OpPrereqError("Please pass at least one modification",
3181 if all_mods.count(True) > 1:
3182 raise errors.OpPrereqError("Can't set the node into more than one"
3183 " state at the same time",
3186 # Boolean value that tells us whether we're offlining or draining the node
3187 self.offline_or_drain = (self.op.offline == True or
3188 self.op.drained == True)
3189 self.deoffline_or_drain = (self.op.offline == False or
3190 self.op.drained == False)
3191 self.might_demote = (self.op.master_candidate == False or
3192 self.offline_or_drain)
3194 self.lock_all = self.op.auto_promote and self.might_demote
3197 def ExpandNames(self):
3199 self.needed_locks = {locking.LEVEL_NODE: locking.ALL_SET}
3201 self.needed_locks = {locking.LEVEL_NODE: self.op.node_name}
3203 def BuildHooksEnv(self):
3206 This runs on the master node.
3210 "OP_TARGET": self.op.node_name,
3211 "MASTER_CANDIDATE": str(self.op.master_candidate),
3212 "OFFLINE": str(self.op.offline),
3213 "DRAINED": str(self.op.drained),
3215 nl = [self.cfg.GetMasterNode(),
3219 def CheckPrereq(self):
3220 """Check prerequisites.
3222 This only checks the instance list against the existing names.
3225 node = self.node = self.cfg.GetNodeInfo(self.op.node_name)
3227 if (self.op.master_candidate is not None or
3228 self.op.drained is not None or
3229 self.op.offline is not None):
3230 # we can't change the master's node flags
3231 if self.op.node_name == self.cfg.GetMasterNode():
3232 raise errors.OpPrereqError("The master role can be changed"
3233 " only via masterfailover",
3237 if node.master_candidate and self.might_demote and not self.lock_all:
3238 assert not self.op.auto_promote, "auto-promote set but lock_all not"
3239 # check if after removing the current node, we're missing master
3241 (mc_remaining, mc_should, _) = \
3242 self.cfg.GetMasterCandidateStats(exceptions=[node.name])
3243 if mc_remaining != mc_should:
3244 raise errors.OpPrereqError("Not enough master candidates, please"
3245 " pass auto_promote to allow promotion",
3248 if (self.op.master_candidate == True and
3249 ((node.offline and not self.op.offline == False) or
3250 (node.drained and not self.op.drained == False))):
3251 raise errors.OpPrereqError("Node '%s' is offline or drained, can't set"
3252 " to master_candidate" % node.name,
3255 # If we're being deofflined/drained, we'll MC ourself if needed
3256 if (self.deoffline_or_drain and not self.offline_or_drain and not
3257 self.op.master_candidate == True and not node.master_candidate):
3258 self.op.master_candidate = _DecideSelfPromotion(self)
3259 if self.op.master_candidate:
3260 self.LogInfo("Autopromoting node to master candidate")
3264 def Exec(self, feedback_fn):
3273 if self.op.offline is not None:
3274 node.offline = self.op.offline
3275 result.append(("offline", str(self.op.offline)))
3276 if self.op.offline == True:
3277 if node.master_candidate:
3278 node.master_candidate = False
3280 result.append(("master_candidate", "auto-demotion due to offline"))
3282 node.drained = False
3283 result.append(("drained", "clear drained status due to offline"))
3285 if self.op.master_candidate is not None:
3286 node.master_candidate = self.op.master_candidate
3288 result.append(("master_candidate", str(self.op.master_candidate)))
3289 if self.op.master_candidate == False:
3290 rrc = self.rpc.call_node_demote_from_mc(node.name)
3293 self.LogWarning("Node failed to demote itself: %s" % msg)
3295 if self.op.drained is not None:
3296 node.drained = self.op.drained
3297 result.append(("drained", str(self.op.drained)))
3298 if self.op.drained == True:
3299 if node.master_candidate:
3300 node.master_candidate = False
3302 result.append(("master_candidate", "auto-demotion due to drain"))
3303 rrc = self.rpc.call_node_demote_from_mc(node.name)
3306 self.LogWarning("Node failed to demote itself: %s" % msg)
3308 node.offline = False
3309 result.append(("offline", "clear offline status due to drain"))
3311 # we locked all nodes, we adjust the CP before updating this node
3313 _AdjustCandidatePool(self, [node.name])
3315 # this will trigger configuration file update, if needed
3316 self.cfg.Update(node, feedback_fn)
3318 # this will trigger job queue propagation or cleanup
3320 self.context.ReaddNode(node)
3325 class LUPowercycleNode(NoHooksLU):
3326 """Powercycles a node.
3329 _OP_REQP = ["node_name", "force"]
3332 def CheckArguments(self):
3333 self.op.node_name = _ExpandNodeName(self.cfg, self.op.node_name)
3334 if self.op.node_name == self.cfg.GetMasterNode() and not self.op.force:
3335 raise errors.OpPrereqError("The node is the master and the force"
3336 " parameter was not set",
3339 def ExpandNames(self):
3340 """Locking for PowercycleNode.
3342 This is a last-resort option and shouldn't block on other
3343 jobs. Therefore, we grab no locks.
3346 self.needed_locks = {}
3348 def CheckPrereq(self):
3349 """Check prerequisites.
3351 This LU has no prereqs.
3356 def Exec(self, feedback_fn):
3360 result = self.rpc.call_node_powercycle(self.op.node_name,
3361 self.cfg.GetHypervisorType())
3362 result.Raise("Failed to schedule the reboot")
3363 return result.payload
3366 class LUQueryClusterInfo(NoHooksLU):
3367 """Query cluster configuration.
3373 def ExpandNames(self):
3374 self.needed_locks = {}
3376 def CheckPrereq(self):
3377 """No prerequsites needed for this LU.
3382 def Exec(self, feedback_fn):
3383 """Return cluster config.
3386 cluster = self.cfg.GetClusterInfo()
3389 # Filter just for enabled hypervisors
3390 for os_name, hv_dict in cluster.os_hvp.items():
3391 os_hvp[os_name] = {}
3392 for hv_name, hv_params in hv_dict.items():
3393 if hv_name in cluster.enabled_hypervisors:
3394 os_hvp[os_name][hv_name] = hv_params
3397 "software_version": constants.RELEASE_VERSION,
3398 "protocol_version": constants.PROTOCOL_VERSION,
3399 "config_version": constants.CONFIG_VERSION,
3400 "os_api_version": max(constants.OS_API_VERSIONS),
3401 "export_version": constants.EXPORT_VERSION,
3402 "architecture": (platform.architecture()[0], platform.machine()),
3403 "name": cluster.cluster_name,
3404 "master": cluster.master_node,
3405 "default_hypervisor": cluster.enabled_hypervisors[0],
3406 "enabled_hypervisors": cluster.enabled_hypervisors,
3407 "hvparams": dict([(hypervisor_name, cluster.hvparams[hypervisor_name])
3408 for hypervisor_name in cluster.enabled_hypervisors]),
3410 "beparams": cluster.beparams,
3411 "nicparams": cluster.nicparams,
3412 "candidate_pool_size": cluster.candidate_pool_size,
3413 "master_netdev": cluster.master_netdev,
3414 "volume_group_name": cluster.volume_group_name,
3415 "file_storage_dir": cluster.file_storage_dir,
3416 "ctime": cluster.ctime,
3417 "mtime": cluster.mtime,
3418 "uuid": cluster.uuid,
3419 "tags": list(cluster.GetTags()),
3425 class LUQueryConfigValues(NoHooksLU):
3426 """Return configuration values.
3431 _FIELDS_DYNAMIC = utils.FieldSet()
3432 _FIELDS_STATIC = utils.FieldSet("cluster_name", "master_node", "drain_flag",
3435 def ExpandNames(self):
3436 self.needed_locks = {}
3438 _CheckOutputFields(static=self._FIELDS_STATIC,
3439 dynamic=self._FIELDS_DYNAMIC,
3440 selected=self.op.output_fields)
3442 def CheckPrereq(self):
3443 """No prerequisites.
3448 def Exec(self, feedback_fn):
3449 """Dump a representation of the cluster config to the standard output.
3453 for field in self.op.output_fields:
3454 if field == "cluster_name":
3455 entry = self.cfg.GetClusterName()
3456 elif field == "master_node":
3457 entry = self.cfg.GetMasterNode()
3458 elif field == "drain_flag":
3459 entry = os.path.exists(constants.JOB_QUEUE_DRAIN_FILE)
3460 elif field == "watcher_pause":
3461 entry = utils.ReadWatcherPauseFile(constants.WATCHER_PAUSEFILE)
3463 raise errors.ParameterError(field)
3464 values.append(entry)
3468 class LUActivateInstanceDisks(NoHooksLU):
3469 """Bring up an instance's disks.
3472 _OP_REQP = ["instance_name"]
3475 def ExpandNames(self):
3476 self._ExpandAndLockInstance()
3477 self.needed_locks[locking.LEVEL_NODE] = []
3478 self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
3480 def DeclareLocks(self, level):
3481 if level == locking.LEVEL_NODE:
3482 self._LockInstancesNodes()
3484 def CheckPrereq(self):
3485 """Check prerequisites.
3487 This checks that the instance is in the cluster.
3490 self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
3491 assert self.instance is not None, \
3492 "Cannot retrieve locked instance %s" % self.op.instance_name
3493 _CheckNodeOnline(self, self.instance.primary_node)
3494 if not hasattr(self.op, "ignore_size"):
3495 self.op.ignore_size = False
3497 def Exec(self, feedback_fn):
3498 """Activate the disks.
3501 disks_ok, disks_info = \
3502 _AssembleInstanceDisks(self, self.instance,
3503 ignore_size=self.op.ignore_size)
3505 raise errors.OpExecError("Cannot activate block devices")
3510 def _AssembleInstanceDisks(lu, instance, ignore_secondaries=False,
3512 """Prepare the block devices for an instance.
3514 This sets up the block devices on all nodes.
3516 @type lu: L{LogicalUnit}
3517 @param lu: the logical unit on whose behalf we execute
3518 @type instance: L{objects.Instance}
3519 @param instance: the instance for whose disks we assemble
3520 @type ignore_secondaries: boolean
3521 @param ignore_secondaries: if true, errors on secondary nodes
3522 won't result in an error return from the function
3523 @type ignore_size: boolean
3524 @param ignore_size: if true, the current known size of the disk
3525 will not be used during the disk activation, useful for cases
3526 when the size is wrong
3527 @return: False if the operation failed, otherwise a list of
3528 (host, instance_visible_name, node_visible_name)
3529 with the mapping from node devices to instance devices
3534 iname = instance.name
3535 # With the two passes mechanism we try to reduce the window of
3536 # opportunity for the race condition of switching DRBD to primary
3537 # before handshaking occured, but we do not eliminate it
3539 # The proper fix would be to wait (with some limits) until the
3540 # connection has been made and drbd transitions from WFConnection
3541 # into any other network-connected state (Connected, SyncTarget,
3544 # 1st pass, assemble on all nodes in secondary mode
3545 for inst_disk in instance.disks:
3546 for node, node_disk in inst_disk.ComputeNodeTree(instance.primary_node):
3548 node_disk = node_disk.Copy()
3549 node_disk.UnsetSize()
3550 lu.cfg.SetDiskID(node_disk, node)
3551 result = lu.rpc.call_blockdev_assemble(node, node_disk, iname, False)
3552 msg = result.fail_msg
3554 lu.proc.LogWarning("Could not prepare block device %s on node %s"
3555 " (is_primary=False, pass=1): %s",
3556 inst_disk.iv_name, node, msg)
3557 if not ignore_secondaries:
3560 # FIXME: race condition on drbd migration to primary
3562 # 2nd pass, do only the primary node
3563 for inst_disk in instance.disks:
3566 for node, node_disk in inst_disk.ComputeNodeTree(instance.primary_node):
3567 if node != instance.primary_node:
3570 node_disk = node_disk.Copy()
3571 node_disk.UnsetSize()
3572 lu.cfg.SetDiskID(node_disk, node)
3573 result = lu.rpc.call_blockdev_assemble(node, node_disk, iname, True)
3574 msg = result.fail_msg
3576 lu.proc.LogWarning("Could not prepare block device %s on node %s"
3577 " (is_primary=True, pass=2): %s",
3578 inst_disk.iv_name, node, msg)
3581 dev_path = result.payload
3583 device_info.append((instance.primary_node, inst_disk.iv_name, dev_path))
3585 # leave the disks configured for the primary node
3586 # this is a workaround that would be fixed better by
3587 # improving the logical/physical id handling
3588 for disk in instance.disks:
3589 lu.cfg.SetDiskID(disk, instance.primary_node)
3591 return disks_ok, device_info
3594 def _StartInstanceDisks(lu, instance, force):
3595 """Start the disks of an instance.
3598 disks_ok, _ = _AssembleInstanceDisks(lu, instance,
3599 ignore_secondaries=force)
3601 _ShutdownInstanceDisks(lu, instance)
3602 if force is not None and not force:
3603 lu.proc.LogWarning("", hint="If the message above refers to a"
3605 " you can retry the operation using '--force'.")
3606 raise errors.OpExecError("Disk consistency error")
3609 class LUDeactivateInstanceDisks(NoHooksLU):
3610 """Shutdown an instance's disks.
3613 _OP_REQP = ["instance_name"]
3616 def ExpandNames(self):
3617 self._ExpandAndLockInstance()
3618 self.needed_locks[locking.LEVEL_NODE] = []
3619 self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
3621 def DeclareLocks(self, level):
3622 if level == locking.LEVEL_NODE:
3623 self._LockInstancesNodes()
3625 def CheckPrereq(self):
3626 """Check prerequisites.
3628 This checks that the instance is in the cluster.
3631 self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
3632 assert self.instance is not None, \
3633 "Cannot retrieve locked instance %s" % self.op.instance_name
3635 def Exec(self, feedback_fn):
3636 """Deactivate the disks
3639 instance = self.instance
3640 _SafeShutdownInstanceDisks(self, instance)
3643 def _SafeShutdownInstanceDisks(lu, instance):
3644 """Shutdown block devices of an instance.
3646 This function checks if an instance is running, before calling
3647 _ShutdownInstanceDisks.
3650 pnode = instance.primary_node
3651 ins_l = lu.rpc.call_instance_list([pnode], [instance.hypervisor])[pnode]
3652 ins_l.Raise("Can't contact node %s" % pnode)
3654 if instance.name in ins_l.payload:
3655 raise errors.OpExecError("Instance is running, can't shutdown"
3658 _ShutdownInstanceDisks(lu, instance)
3661 def _ShutdownInstanceDisks(lu, instance, ignore_primary=False):
3662 """Shutdown block devices of an instance.
3664 This does the shutdown on all nodes of the instance.
3666 If the ignore_primary is false, errors on the primary node are
3671 for disk in instance.disks:
3672 for node, top_disk in disk.ComputeNodeTree(instance.primary_node):
3673 lu.cfg.SetDiskID(top_disk, node)
3674 result = lu.rpc.call_blockdev_shutdown(node, top_disk)
3675 msg = result.fail_msg
3677 lu.LogWarning("Could not shutdown block device %s on node %s: %s",
3678 disk.iv_name, node, msg)
3679 if not ignore_primary or node != instance.primary_node:
3684 def _CheckNodeFreeMemory(lu, node, reason, requested, hypervisor_name):
3685 """Checks if a node has enough free memory.
3687 This function check if a given node has the needed amount of free
3688 memory. In case the node has less memory or we cannot get the
3689 information from the node, this function raise an OpPrereqError
3692 @type lu: C{LogicalUnit}
3693 @param lu: a logical unit from which we get configuration data
3695 @param node: the node to check
3696 @type reason: C{str}
3697 @param reason: string to use in the error message
3698 @type requested: C{int}
3699 @param requested: the amount of memory in MiB to check for
3700 @type hypervisor_name: C{str}
3701 @param hypervisor_name: the hypervisor to ask for memory stats
3702 @raise errors.OpPrereqError: if the node doesn't have enough memory, or
3703 we cannot check the node
3706 nodeinfo = lu.rpc.call_node_info([node], lu.cfg.GetVGName(), hypervisor_name)
3707 nodeinfo[node].Raise("Can't get data from node %s" % node,
3708 prereq=True, ecode=errors.ECODE_ENVIRON)
3709 free_mem = nodeinfo[node].payload.get('memory_free', None)
3710 if not isinstance(free_mem, int):
3711 raise errors.OpPrereqError("Can't compute free memory on node %s, result"
3712 " was '%s'" % (node, free_mem),
3713 errors.ECODE_ENVIRON)
3714 if requested > free_mem:
3715 raise errors.OpPrereqError("Not enough memory on node %s for %s:"
3716 " needed %s MiB, available %s MiB" %
3717 (node, reason, requested, free_mem),
3721 class LUStartupInstance(LogicalUnit):
3722 """Starts an instance.
3725 HPATH = "instance-start"
3726 HTYPE = constants.HTYPE_INSTANCE
3727 _OP_REQP = ["instance_name", "force"]
3730 def ExpandNames(self):
3731 self._ExpandAndLockInstance()
3733 def BuildHooksEnv(self):
3736 This runs on master, primary and secondary nodes of the instance.
3740 "FORCE": self.op.force,
3742 env.update(_BuildInstanceHookEnvByObject(self, self.instance))
3743 nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
3746 def CheckPrereq(self):
3747 """Check prerequisites.
3749 This checks that the instance is in the cluster.
3752 self.instance = instance = self.cfg.GetInstanceInfo(self.op.instance_name)
3753 assert self.instance is not None, \
3754 "Cannot retrieve locked instance %s" % self.op.instance_name
3757 self.beparams = getattr(self.op, "beparams", {})
3759 if not isinstance(self.beparams, dict):
3760 raise errors.OpPrereqError("Invalid beparams passed: %s, expected"
3761 " dict" % (type(self.beparams), ),
3763 # fill the beparams dict
3764 utils.ForceDictType(self.beparams, constants.BES_PARAMETER_TYPES)
3765 self.op.beparams = self.beparams
3768 self.hvparams = getattr(self.op, "hvparams", {})
3770 if not isinstance(self.hvparams, dict):
3771 raise errors.OpPrereqError("Invalid hvparams passed: %s, expected"
3772 " dict" % (type(self.hvparams), ),
3775 # check hypervisor parameter syntax (locally)
3776 cluster = self.cfg.GetClusterInfo()
3777 utils.ForceDictType(self.hvparams, constants.HVS_PARAMETER_TYPES)
3778 filled_hvp = objects.FillDict(cluster.hvparams[instance.hypervisor],
3780 filled_hvp.update(self.hvparams)
3781 hv_type = hypervisor.GetHypervisor(instance.hypervisor)
3782 hv_type.CheckParameterSyntax(filled_hvp)
3783 _CheckHVParams(self, instance.all_nodes, instance.hypervisor, filled_hvp)
3784 self.op.hvparams = self.hvparams
3786 _CheckNodeOnline(self, instance.primary_node)
3788 bep = self.cfg.GetClusterInfo().FillBE(instance)
3789 # check bridges existence
3790 _CheckInstanceBridgesExist(self, instance)
3792 remote_info = self.rpc.call_instance_info(instance.primary_node,
3794 instance.hypervisor)
3795 remote_info.Raise("Error checking node %s" % instance.primary_node,
3796 prereq=True, ecode=errors.ECODE_ENVIRON)
3797 if not remote_info.payload: # not running already
3798 _CheckNodeFreeMemory(self, instance.primary_node,
3799 "starting instance %s" % instance.name,
3800 bep[constants.BE_MEMORY], instance.hypervisor)
3802 def Exec(self, feedback_fn):
3803 """Start the instance.
3806 instance = self.instance
3807 force = self.op.force
3809 self.cfg.MarkInstanceUp(instance.name)
3811 node_current = instance.primary_node
3813 _StartInstanceDisks(self, instance, force)
3815 result = self.rpc.call_instance_start(node_current, instance,
3816 self.hvparams, self.beparams)
3817 msg = result.fail_msg
3819 _ShutdownInstanceDisks(self, instance)
3820 raise errors.OpExecError("Could not start instance: %s" % msg)
3823 class LURebootInstance(LogicalUnit):
3824 """Reboot an instance.
3827 HPATH = "instance-reboot"
3828 HTYPE = constants.HTYPE_INSTANCE
3829 _OP_REQP = ["instance_name", "ignore_secondaries", "reboot_type"]
3832 def CheckArguments(self):
3833 """Check the arguments.
3836 self.shutdown_timeout = getattr(self.op, "shutdown_timeout",
3837 constants.DEFAULT_SHUTDOWN_TIMEOUT)
3839 def ExpandNames(self):
3840 if self.op.reboot_type not in [constants.INSTANCE_REBOOT_SOFT,
3841 constants.INSTANCE_REBOOT_HARD,
3842 constants.INSTANCE_REBOOT_FULL]:
3843 raise errors.ParameterError("reboot type not in [%s, %s, %s]" %
3844 (constants.INSTANCE_REBOOT_SOFT,
3845 constants.INSTANCE_REBOOT_HARD,
3846 constants.INSTANCE_REBOOT_FULL))
3847 self._ExpandAndLockInstance()
3849 def BuildHooksEnv(self):
3852 This runs on master, primary and secondary nodes of the instance.
3856 "IGNORE_SECONDARIES": self.op.ignore_secondaries,
3857 "REBOOT_TYPE": self.op.reboot_type,
3858 "SHUTDOWN_TIMEOUT": self.shutdown_timeout,
3860 env.update(_BuildInstanceHookEnvByObject(self, self.instance))
3861 nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
3864 def CheckPrereq(self):
3865 """Check prerequisites.
3867 This checks that the instance is in the cluster.
3870 self.instance = instance = self.cfg.GetInstanceInfo(self.op.instance_name)
3871 assert self.instance is not None, \
3872 "Cannot retrieve locked instance %s" % self.op.instance_name
3874 _CheckNodeOnline(self, instance.primary_node)
3876 # check bridges existence
3877 _CheckInstanceBridgesExist(self, instance)
3879 def Exec(self, feedback_fn):
3880 """Reboot the instance.
3883 instance = self.instance
3884 ignore_secondaries = self.op.ignore_secondaries
3885 reboot_type = self.op.reboot_type
3887 node_current = instance.primary_node
3889 if reboot_type in [constants.INSTANCE_REBOOT_SOFT,
3890 constants.INSTANCE_REBOOT_HARD]:
3891 for disk in instance.disks:
3892 self.cfg.SetDiskID(disk, node_current)
3893 result = self.rpc.call_instance_reboot(node_current, instance,
3895 self.shutdown_timeout)
3896 result.Raise("Could not reboot instance")
3898 result = self.rpc.call_instance_shutdown(node_current, instance,
3899 self.shutdown_timeout)
3900 result.Raise("Could not shutdown instance for full reboot")
3901 _ShutdownInstanceDisks(self, instance)
3902 _StartInstanceDisks(self, instance, ignore_secondaries)
3903 result = self.rpc.call_instance_start(node_current, instance, None, None)
3904 msg = result.fail_msg
3906 _ShutdownInstanceDisks(self, instance)
3907 raise errors.OpExecError("Could not start instance for"
3908 " full reboot: %s" % msg)
3910 self.cfg.MarkInstanceUp(instance.name)
3913 class LUShutdownInstance(LogicalUnit):
3914 """Shutdown an instance.
3917 HPATH = "instance-stop"
3918 HTYPE = constants.HTYPE_INSTANCE
3919 _OP_REQP = ["instance_name"]
3922 def CheckArguments(self):
3923 """Check the arguments.
3926 self.timeout = getattr(self.op, "timeout",
3927 constants.DEFAULT_SHUTDOWN_TIMEOUT)
3929 def ExpandNames(self):
3930 self._ExpandAndLockInstance()
3932 def BuildHooksEnv(self):
3935 This runs on master, primary and secondary nodes of the instance.
3938 env = _BuildInstanceHookEnvByObject(self, self.instance)
3939 env["TIMEOUT"] = self.timeout
3940 nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
3943 def CheckPrereq(self):
3944 """Check prerequisites.
3946 This checks that the instance is in the cluster.
3949 self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
3950 assert self.instance is not None, \
3951 "Cannot retrieve locked instance %s" % self.op.instance_name
3952 _CheckNodeOnline(self, self.instance.primary_node)
3954 def Exec(self, feedback_fn):
3955 """Shutdown the instance.
3958 instance = self.instance
3959 node_current = instance.primary_node
3960 timeout = self.timeout
3961 self.cfg.MarkInstanceDown(instance.name)
3962 result = self.rpc.call_instance_shutdown(node_current, instance, timeout)
3963 msg = result.fail_msg
3965 self.proc.LogWarning("Could not shutdown instance: %s" % msg)
3967 _ShutdownInstanceDisks(self, instance)
3970 class LUReinstallInstance(LogicalUnit):
3971 """Reinstall an instance.
3974 HPATH = "instance-reinstall"
3975 HTYPE = constants.HTYPE_INSTANCE
3976 _OP_REQP = ["instance_name"]
3979 def ExpandNames(self):
3980 self._ExpandAndLockInstance()
3982 def BuildHooksEnv(self):
3985 This runs on master, primary and secondary nodes of the instance.
3988 env = _BuildInstanceHookEnvByObject(self, self.instance)
3989 nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
3992 def CheckPrereq(self):
3993 """Check prerequisites.
3995 This checks that the instance is in the cluster and is not running.
3998 instance = self.cfg.GetInstanceInfo(self.op.instance_name)
3999 assert instance is not None, \
4000 "Cannot retrieve locked instance %s" % self.op.instance_name
4001 _CheckNodeOnline(self, instance.primary_node)
4003 if instance.disk_template == constants.DT_DISKLESS:
4004 raise errors.OpPrereqError("Instance '%s' has no disks" %
4005 self.op.instance_name,
4007 if instance.admin_up:
4008 raise errors.OpPrereqError("Instance '%s' is marked to be up" %
4009 self.op.instance_name,
4011 remote_info = self.rpc.call_instance_info(instance.primary_node,
4013 instance.hypervisor)
4014 remote_info.Raise("Error checking node %s" % instance.primary_node,
4015 prereq=True, ecode=errors.ECODE_ENVIRON)
4016 if remote_info.payload:
4017 raise errors.OpPrereqError("Instance '%s' is running on the node %s" %
4018 (self.op.instance_name,
4019 instance.primary_node),
4022 self.op.os_type = getattr(self.op, "os_type", None)
4023 self.op.force_variant = getattr(self.op, "force_variant", False)
4024 if self.op.os_type is not None:
4026 pnode = _ExpandNodeName(self.cfg, instance.primary_node)
4027 result = self.rpc.call_os_get(pnode, self.op.os_type)
4028 result.Raise("OS '%s' not in supported OS list for primary node %s" %
4029 (self.op.os_type, pnode),
4030 prereq=True, ecode=errors.ECODE_INVAL)
4031 if not self.op.force_variant:
4032 _CheckOSVariant(result.payload, self.op.os_type)
4034 self.instance = instance
4036 def Exec(self, feedback_fn):
4037 """Reinstall the instance.
4040 inst = self.instance
4042 if self.op.os_type is not None:
4043 feedback_fn("Changing OS to '%s'..." % self.op.os_type)
4044 inst.os = self.op.os_type
4045 self.cfg.Update(inst, feedback_fn)
4047 _StartInstanceDisks(self, inst, None)
4049 feedback_fn("Running the instance OS create scripts...")
4050 # FIXME: pass debug option from opcode to backend
4051 result = self.rpc.call_instance_os_add(inst.primary_node, inst, True,
4052 self.op.debug_level)
4053 result.Raise("Could not install OS for instance %s on node %s" %
4054 (inst.name, inst.primary_node))
4056 _ShutdownInstanceDisks(self, inst)
4059 class LURecreateInstanceDisks(LogicalUnit):
4060 """Recreate an instance's missing disks.
4063 HPATH = "instance-recreate-disks"
4064 HTYPE = constants.HTYPE_INSTANCE
4065 _OP_REQP = ["instance_name", "disks"]
4068 def CheckArguments(self):
4069 """Check the arguments.
4072 if not isinstance(self.op.disks, list):
4073 raise errors.OpPrereqError("Invalid disks parameter", errors.ECODE_INVAL)
4074 for item in self.op.disks:
4075 if (not isinstance(item, int) or
4077 raise errors.OpPrereqError("Invalid disk specification '%s'" %
4078 str(item), errors.ECODE_INVAL)
4080 def ExpandNames(self):
4081 self._ExpandAndLockInstance()
4083 def BuildHooksEnv(self):
4086 This runs on master, primary and secondary nodes of the instance.
4089 env = _BuildInstanceHookEnvByObject(self, self.instance)
4090 nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
4093 def CheckPrereq(self):
4094 """Check prerequisites.
4096 This checks that the instance is in the cluster and is not running.
4099 instance = self.cfg.GetInstanceInfo(self.op.instance_name)
4100 assert instance is not None, \
4101 "Cannot retrieve locked instance %s" % self.op.instance_name
4102 _CheckNodeOnline(self, instance.primary_node)
4104 if instance.disk_template == constants.DT_DISKLESS:
4105 raise errors.OpPrereqError("Instance '%s' has no disks" %
4106 self.op.instance_name, errors.ECODE_INVAL)
4107 if instance.admin_up:
4108 raise errors.OpPrereqError("Instance '%s' is marked to be up" %
4109 self.op.instance_name, errors.ECODE_STATE)
4110 remote_info = self.rpc.call_instance_info(instance.primary_node,
4112 instance.hypervisor)
4113 remote_info.Raise("Error checking node %s" % instance.primary_node,
4114 prereq=True, ecode=errors.ECODE_ENVIRON)
4115 if remote_info.payload:
4116 raise errors.OpPrereqError("Instance '%s' is running on the node %s" %
4117 (self.op.instance_name,
4118 instance.primary_node), errors.ECODE_STATE)
4120 if not self.op.disks:
4121 self.op.disks = range(len(instance.disks))
4123 for idx in self.op.disks:
4124 if idx >= len(instance.disks):
4125 raise errors.OpPrereqError("Invalid disk index passed '%s'" % idx,
4128 self.instance = instance
4130 def Exec(self, feedback_fn):
4131 """Recreate the disks.
4135 for idx, _ in enumerate(self.instance.disks):
4136 if idx not in self.op.disks: # disk idx has not been passed in
4140 _CreateDisks(self, self.instance, to_skip=to_skip)
4143 class LURenameInstance(LogicalUnit):
4144 """Rename an instance.
4147 HPATH = "instance-rename"
4148 HTYPE = constants.HTYPE_INSTANCE
4149 _OP_REQP = ["instance_name", "new_name"]
4151 def BuildHooksEnv(self):
4154 This runs on master, primary and secondary nodes of the instance.
4157 env = _BuildInstanceHookEnvByObject(self, self.instance)
4158 env["INSTANCE_NEW_NAME"] = self.op.new_name
4159 nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
4162 def CheckPrereq(self):
4163 """Check prerequisites.
4165 This checks that the instance is in the cluster and is not running.
4168 self.op.instance_name = _ExpandInstanceName(self.cfg,
4169 self.op.instance_name)
4170 instance = self.cfg.GetInstanceInfo(self.op.instance_name)
4171 assert instance is not None
4172 _CheckNodeOnline(self, instance.primary_node)
4174 if instance.admin_up:
4175 raise errors.OpPrereqError("Instance '%s' is marked to be up" %
4176 self.op.instance_name, errors.ECODE_STATE)
4177 remote_info = self.rpc.call_instance_info(instance.primary_node,
4179 instance.hypervisor)
4180 remote_info.Raise("Error checking node %s" % instance.primary_node,
4181 prereq=True, ecode=errors.ECODE_ENVIRON)
4182 if remote_info.payload:
4183 raise errors.OpPrereqError("Instance '%s' is running on the node %s" %
4184 (self.op.instance_name,
4185 instance.primary_node), errors.ECODE_STATE)
4186 self.instance = instance
4188 # new name verification
4189 name_info = utils.GetHostInfo(self.op.new_name)
4191 self.op.new_name = new_name = name_info.name
4192 instance_list = self.cfg.GetInstanceList()
4193 if new_name in instance_list:
4194 raise errors.OpPrereqError("Instance '%s' is already in the cluster" %
4195 new_name, errors.ECODE_EXISTS)
4197 if not getattr(self.op, "ignore_ip", False):
4198 if utils.TcpPing(name_info.ip, constants.DEFAULT_NODED_PORT):
4199 raise errors.OpPrereqError("IP %s of instance %s already in use" %
4200 (name_info.ip, new_name),
4201 errors.ECODE_NOTUNIQUE)
4204 def Exec(self, feedback_fn):
4205 """Reinstall the instance.
4208 inst = self.instance
4209 old_name = inst.name
4211 if inst.disk_template == constants.DT_FILE:
4212 old_file_storage_dir = os.path.dirname(inst.disks[0].logical_id[1])
4214 self.cfg.RenameInstance(inst.name, self.op.new_name)
4215 # Change the instance lock. This is definitely safe while we hold the BGL
4216 self.context.glm.remove(locking.LEVEL_INSTANCE, old_name)
4217 self.context.glm.add(locking.LEVEL_INSTANCE, self.op.new_name)
4219 # re-read the instance from the configuration after rename
4220 inst = self.cfg.GetInstanceInfo(self.op.new_name)
4222 if inst.disk_template == constants.DT_FILE:
4223 new_file_storage_dir = os.path.dirname(inst.disks[0].logical_id[1])
4224 result = self.rpc.call_file_storage_dir_rename(inst.primary_node,
4225 old_file_storage_dir,
4226 new_file_storage_dir)
4227 result.Raise("Could not rename on node %s directory '%s' to '%s'"
4228 " (but the instance has been renamed in Ganeti)" %
4229 (inst.primary_node, old_file_storage_dir,
4230 new_file_storage_dir))
4232 _StartInstanceDisks(self, inst, None)
4234 result = self.rpc.call_instance_run_rename(inst.primary_node, inst,
4235 old_name, self.op.debug_level)
4236 msg = result.fail_msg
4238 msg = ("Could not run OS rename script for instance %s on node %s"
4239 " (but the instance has been renamed in Ganeti): %s" %
4240 (inst.name, inst.primary_node, msg))
4241 self.proc.LogWarning(msg)
4243 _ShutdownInstanceDisks(self, inst)
4246 class LURemoveInstance(LogicalUnit):
4247 """Remove an instance.
4250 HPATH = "instance-remove"
4251 HTYPE = constants.HTYPE_INSTANCE
4252 _OP_REQP = ["instance_name", "ignore_failures"]
4255 def CheckArguments(self):
4256 """Check the arguments.
4259 self.shutdown_timeout = getattr(self.op, "shutdown_timeout",
4260 constants.DEFAULT_SHUTDOWN_TIMEOUT)
4262 def ExpandNames(self):
4263 self._ExpandAndLockInstance()
4264 self.needed_locks[locking.LEVEL_NODE] = []
4265 self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
4267 def DeclareLocks(self, level):
4268 if level == locking.LEVEL_NODE:
4269 self._LockInstancesNodes()
4271 def BuildHooksEnv(self):
4274 This runs on master, primary and secondary nodes of the instance.
4277 env = _BuildInstanceHookEnvByObject(self, self.instance)
4278 env["SHUTDOWN_TIMEOUT"] = self.shutdown_timeout
4279 nl = [self.cfg.GetMasterNode()]
4280 nl_post = list(self.instance.all_nodes) + nl
4281 return env, nl, nl_post
4283 def CheckPrereq(self):
4284 """Check prerequisites.
4286 This checks that the instance is in the cluster.
4289 self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
4290 assert self.instance is not None, \
4291 "Cannot retrieve locked instance %s" % self.op.instance_name
4293 def Exec(self, feedback_fn):
4294 """Remove the instance.
4297 instance = self.instance
4298 logging.info("Shutting down instance %s on node %s",
4299 instance.name, instance.primary_node)
4301 result = self.rpc.call_instance_shutdown(instance.primary_node, instance,
4302 self.shutdown_timeout)
4303 msg = result.fail_msg
4305 if self.op.ignore_failures:
4306 feedback_fn("Warning: can't shutdown instance: %s" % msg)
4308 raise errors.OpExecError("Could not shutdown instance %s on"
4310 (instance.name, instance.primary_node, msg))
4312 logging.info("Removing block devices for instance %s", instance.name)
4314 if not _RemoveDisks(self, instance):
4315 if self.op.ignore_failures:
4316 feedback_fn("Warning: can't remove instance's disks")
4318 raise errors.OpExecError("Can't remove instance's disks")
4320 logging.info("Removing instance %s out of cluster config", instance.name)
4322 self.cfg.RemoveInstance(instance.name)
4323 self.remove_locks[locking.LEVEL_INSTANCE] = instance.name
4326 class LUQueryInstances(NoHooksLU):
4327 """Logical unit for querying instances.
4330 # pylint: disable-msg=W0142
4331 _OP_REQP = ["output_fields", "names", "use_locking"]
4333 _SIMPLE_FIELDS = ["name", "os", "network_port", "hypervisor",
4334 "serial_no", "ctime", "mtime", "uuid"]
4335 _FIELDS_STATIC = utils.FieldSet(*["name", "os", "pnode", "snodes",
4337 "disk_template", "ip", "mac", "bridge",
4338 "nic_mode", "nic_link",
4339 "sda_size", "sdb_size", "vcpus", "tags",
4340 "network_port", "beparams",
4341 r"(disk)\.(size)/([0-9]+)",
4342 r"(disk)\.(sizes)", "disk_usage",
4343 r"(nic)\.(mac|ip|mode|link)/([0-9]+)",
4344 r"(nic)\.(bridge)/([0-9]+)",
4345 r"(nic)\.(macs|ips|modes|links|bridges)",
4346 r"(disk|nic)\.(count)",
4348 ] + _SIMPLE_FIELDS +
4350 for name in constants.HVS_PARAMETERS
4351 if name not in constants.HVC_GLOBALS] +
4353 for name in constants.BES_PARAMETERS])
4354 _FIELDS_DYNAMIC = utils.FieldSet("oper_state", "oper_ram", "status")
4357 def ExpandNames(self):
4358 _CheckOutputFields(static=self._FIELDS_STATIC,
4359 dynamic=self._FIELDS_DYNAMIC,
4360 selected=self.op.output_fields)
4362 self.needed_locks = {}
4363 self.share_locks[locking.LEVEL_INSTANCE] = 1
4364 self.share_locks[locking.LEVEL_NODE] = 1
4367 self.wanted = _GetWantedInstances(self, self.op.names)
4369 self.wanted = locking.ALL_SET
4371 self.do_node_query = self._FIELDS_STATIC.NonMatching(self.op.output_fields)
4372 self.do_locking = self.do_node_query and self.op.use_locking
4374 self.needed_locks[locking.LEVEL_INSTANCE] = self.wanted
4375 self.needed_locks[locking.LEVEL_NODE] = []
4376 self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
4378 def DeclareLocks(self, level):
4379 if level == locking.LEVEL_NODE and self.do_locking:
4380 self._LockInstancesNodes()
4382 def CheckPrereq(self):
4383 """Check prerequisites.
4388 def Exec(self, feedback_fn):
4389 """Computes the list of nodes and their attributes.
4392 # pylint: disable-msg=R0912
4393 # way too many branches here
4394 all_info = self.cfg.GetAllInstancesInfo()
4395 if self.wanted == locking.ALL_SET:
4396 # caller didn't specify instance names, so ordering is not important
4398 instance_names = self.acquired_locks[locking.LEVEL_INSTANCE]
4400 instance_names = all_info.keys()
4401 instance_names = utils.NiceSort(instance_names)
4403 # caller did specify names, so we must keep the ordering
4405 tgt_set = self.acquired_locks[locking.LEVEL_INSTANCE]
4407 tgt_set = all_info.keys()
4408 missing = set(self.wanted).difference(tgt_set)
4410 raise errors.OpExecError("Some instances were removed before"
4411 " retrieving their data: %s" % missing)
4412 instance_names = self.wanted
4414 instance_list = [all_info[iname] for iname in instance_names]
4416 # begin data gathering
4418 nodes = frozenset([inst.primary_node for inst in instance_list])
4419 hv_list = list(set([inst.hypervisor for inst in instance_list]))
4423 if self.do_node_query:
4425 node_data = self.rpc.call_all_instances_info(nodes, hv_list)
4427 result = node_data[name]
4429 # offline nodes will be in both lists
4430 off_nodes.append(name)
4432 bad_nodes.append(name)
4435 live_data.update(result.payload)
4436 # else no instance is alive
4438 live_data = dict([(name, {}) for name in instance_names])
4440 # end data gathering
4445 cluster = self.cfg.GetClusterInfo()
4446 for instance in instance_list:
4448 i_hv = cluster.FillHV(instance, skip_globals=True)
4449 i_be = cluster.FillBE(instance)
4450 i_nicp = [objects.FillDict(cluster.nicparams[constants.PP_DEFAULT],
4451 nic.nicparams) for nic in instance.nics]
4452 for field in self.op.output_fields:
4453 st_match = self._FIELDS_STATIC.Matches(field)
4454 if field in self._SIMPLE_FIELDS:
4455 val = getattr(instance, field)
4456 elif field == "pnode":
4457 val = instance.primary_node
4458 elif field == "snodes":
4459 val = list(instance.secondary_nodes)
4460 elif field == "admin_state":
4461 val = instance.admin_up
4462 elif field == "oper_state":
4463 if instance.primary_node in bad_nodes:
4466 val = bool(live_data.get(instance.name))
4467 elif field == "status":
4468 if instance.primary_node in off_nodes:
4469 val = "ERROR_nodeoffline"
4470 elif instance.primary_node in bad_nodes:
4471 val = "ERROR_nodedown"
4473 running = bool(live_data.get(instance.name))
4475 if instance.admin_up:
4480 if instance.admin_up:
4484 elif field == "oper_ram":
4485 if instance.primary_node in bad_nodes:
4487 elif instance.name in live_data:
4488 val = live_data[instance.name].get("memory", "?")
4491 elif field == "vcpus":
4492 val = i_be[constants.BE_VCPUS]
4493 elif field == "disk_template":
4494 val = instance.disk_template
4497 val = instance.nics[0].ip
4500 elif field == "nic_mode":
4502 val = i_nicp[0][constants.NIC_MODE]
4505 elif field == "nic_link":
4507 val = i_nicp[0][constants.NIC_LINK]
4510 elif field == "bridge":
4511 if (instance.nics and
4512 i_nicp[0][constants.NIC_MODE] == constants.NIC_MODE_BRIDGED):
4513 val = i_nicp[0][constants.NIC_LINK]
4516 elif field == "mac":
4518 val = instance.nics[0].mac
4521 elif field == "sda_size" or field == "sdb_size":
4522 idx = ord(field[2]) - ord('a')
4524 val = instance.FindDisk(idx).size
4525 except errors.OpPrereqError:
4527 elif field == "disk_usage": # total disk usage per node
4528 disk_sizes = [{'size': disk.size} for disk in instance.disks]
4529 val = _ComputeDiskSize(instance.disk_template, disk_sizes)
4530 elif field == "tags":
4531 val = list(instance.GetTags())
4532 elif field == "hvparams":
4534 elif (field.startswith(HVPREFIX) and
4535 field[len(HVPREFIX):] in constants.HVS_PARAMETERS and
4536 field[len(HVPREFIX):] not in constants.HVC_GLOBALS):
4537 val = i_hv.get(field[len(HVPREFIX):], None)
4538 elif field == "beparams":
4540 elif (field.startswith(BEPREFIX) and
4541 field[len(BEPREFIX):] in constants.BES_PARAMETERS):
4542 val = i_be.get(field[len(BEPREFIX):], None)
4543 elif st_match and st_match.groups():
4544 # matches a variable list
4545 st_groups = st_match.groups()
4546 if st_groups and st_groups[0] == "disk":
4547 if st_groups[1] == "count":
4548 val = len(instance.disks)
4549 elif st_groups[1] == "sizes":
4550 val = [disk.size for disk in instance.disks]
4551 elif st_groups[1] == "size":
4553 val = instance.FindDisk(st_groups[2]).size
4554 except errors.OpPrereqError:
4557 assert False, "Unhandled disk parameter"
4558 elif st_groups[0] == "nic":
4559 if st_groups[1] == "count":
4560 val = len(instance.nics)
4561 elif st_groups[1] == "macs":
4562 val = [nic.mac for nic in instance.nics]
4563 elif st_groups[1] == "ips":
4564 val = [nic.ip for nic in instance.nics]
4565 elif st_groups[1] == "modes":
4566 val = [nicp[constants.NIC_MODE] for nicp in i_nicp]
4567 elif st_groups[1] == "links":
4568 val = [nicp[constants.NIC_LINK] for nicp in i_nicp]
4569 elif st_groups[1] == "bridges":
4572 if nicp[constants.NIC_MODE] == constants.NIC_MODE_BRIDGED:
4573 val.append(nicp[constants.NIC_LINK])
4578 nic_idx = int(st_groups[2])
4579 if nic_idx >= len(instance.nics):
4582 if st_groups[1] == "mac":
4583 val = instance.nics[nic_idx].mac
4584 elif st_groups[1] == "ip":
4585 val = instance.nics[nic_idx].ip
4586 elif st_groups[1] == "mode":
4587 val = i_nicp[nic_idx][constants.NIC_MODE]
4588 elif st_groups[1] == "link":
4589 val = i_nicp[nic_idx][constants.NIC_LINK]
4590 elif st_groups[1] == "bridge":
4591 nic_mode = i_nicp[nic_idx][constants.NIC_MODE]
4592 if nic_mode == constants.NIC_MODE_BRIDGED:
4593 val = i_nicp[nic_idx][constants.NIC_LINK]
4597 assert False, "Unhandled NIC parameter"
4599 assert False, ("Declared but unhandled variable parameter '%s'" %
4602 assert False, "Declared but unhandled parameter '%s'" % field
4609 class LUFailoverInstance(LogicalUnit):
4610 """Failover an instance.
4613 HPATH = "instance-failover"
4614 HTYPE = constants.HTYPE_INSTANCE
4615 _OP_REQP = ["instance_name", "ignore_consistency"]
4618 def CheckArguments(self):
4619 """Check the arguments.
4622 self.shutdown_timeout = getattr(self.op, "shutdown_timeout",
4623 constants.DEFAULT_SHUTDOWN_TIMEOUT)
4625 def ExpandNames(self):
4626 self._ExpandAndLockInstance()
4627 self.needed_locks[locking.LEVEL_NODE] = []
4628 self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
4630 def DeclareLocks(self, level):
4631 if level == locking.LEVEL_NODE:
4632 self._LockInstancesNodes()
4634 def BuildHooksEnv(self):
4637 This runs on master, primary and secondary nodes of the instance.
4640 instance = self.instance
4641 source_node = instance.primary_node
4642 target_node = instance.secondary_nodes[0]
4644 "IGNORE_CONSISTENCY": self.op.ignore_consistency,
4645 "SHUTDOWN_TIMEOUT": self.shutdown_timeout,
4646 "OLD_PRIMARY": source_node,
4647 "OLD_SECONDARY": target_node,
4648 "NEW_PRIMARY": target_node,
4649 "NEW_SECONDARY": source_node,
4651 env.update(_BuildInstanceHookEnvByObject(self, instance))
4652 nl = [self.cfg.GetMasterNode()] + list(instance.secondary_nodes)
4654 nl_post.append(source_node)
4655 return env, nl, nl_post
4657 def CheckPrereq(self):
4658 """Check prerequisites.
4660 This checks that the instance is in the cluster.
4663 self.instance = instance = self.cfg.GetInstanceInfo(self.op.instance_name)
4664 assert self.instance is not None, \
4665 "Cannot retrieve locked instance %s" % self.op.instance_name
4667 bep = self.cfg.GetClusterInfo().FillBE(instance)
4668 if instance.disk_template not in constants.DTS_NET_MIRROR:
4669 raise errors.OpPrereqError("Instance's disk layout is not"
4670 " network mirrored, cannot failover.",
4673 secondary_nodes = instance.secondary_nodes
4674 if not secondary_nodes:
4675 raise errors.ProgrammerError("no secondary node but using "
4676 "a mirrored disk template")
4678 target_node = secondary_nodes[0]
4679 _CheckNodeOnline(self, target_node)
4680 _CheckNodeNotDrained(self, target_node)
4681 if instance.admin_up:
4682 # check memory requirements on the secondary node
4683 _CheckNodeFreeMemory(self, target_node, "failing over instance %s" %
4684 instance.name, bep[constants.BE_MEMORY],
4685 instance.hypervisor)
4687 self.LogInfo("Not checking memory on the secondary node as"
4688 " instance will not be started")
4690 # check bridge existance
4691 _CheckInstanceBridgesExist(self, instance, node=target_node)
4693 def Exec(self, feedback_fn):
4694 """Failover an instance.
4696 The failover is done by shutting it down on its present node and
4697 starting it on the secondary.
4700 instance = self.instance
4702 source_node = instance.primary_node
4703 target_node = instance.secondary_nodes[0]
4705 if instance.admin_up:
4706 feedback_fn("* checking disk consistency between source and target")
4707 for dev in instance.disks:
4708 # for drbd, these are drbd over lvm
4709 if not _CheckDiskConsistency(self, dev, target_node, False):
4710 if not self.op.ignore_consistency:
4711 raise errors.OpExecError("Disk %s is degraded on target node,"
4712 " aborting failover." % dev.iv_name)
4714 feedback_fn("* not checking disk consistency as instance is not running")
4716 feedback_fn("* shutting down instance on source node")
4717 logging.info("Shutting down instance %s on node %s",
4718 instance.name, source_node)
4720 result = self.rpc.call_instance_shutdown(source_node, instance,
4721 self.shutdown_timeout)
4722 msg = result.fail_msg
4724 if self.op.ignore_consistency:
4725 self.proc.LogWarning("Could not shutdown instance %s on node %s."
4726 " Proceeding anyway. Please make sure node"
4727 " %s is down. Error details: %s",
4728 instance.name, source_node, source_node, msg)
4730 raise errors.OpExecError("Could not shutdown instance %s on"
4732 (instance.name, source_node, msg))
4734 feedback_fn("* deactivating the instance's disks on source node")
4735 if not _ShutdownInstanceDisks(self, instance, ignore_primary=True):
4736 raise errors.OpExecError("Can't shut down the instance's disks.")
4738 instance.primary_node = target_node
4739 # distribute new instance config to the other nodes
4740 self.cfg.Update(instance, feedback_fn)
4742 # Only start the instance if it's marked as up
4743 if instance.admin_up:
4744 feedback_fn("* activating the instance's disks on target node")
4745 logging.info("Starting instance %s on node %s",
4746 instance.name, target_node)
4748 disks_ok, _ = _AssembleInstanceDisks(self, instance,
4749 ignore_secondaries=True)
4751 _ShutdownInstanceDisks(self, instance)
4752 raise errors.OpExecError("Can't activate the instance's disks")
4754 feedback_fn("* starting the instance on the target node")
4755 result = self.rpc.call_instance_start(target_node, instance, None, None)
4756 msg = result.fail_msg
4758 _ShutdownInstanceDisks(self, instance)
4759 raise errors.OpExecError("Could not start instance %s on node %s: %s" %
4760 (instance.name, target_node, msg))
4763 class LUMigrateInstance(LogicalUnit):
4764 """Migrate an instance.
4766 This is migration without shutting down, compared to the failover,
4767 which is done with shutdown.
4770 HPATH = "instance-migrate"
4771 HTYPE = constants.HTYPE_INSTANCE
4772 _OP_REQP = ["instance_name", "live", "cleanup"]
4776 def ExpandNames(self):
4777 self._ExpandAndLockInstance()
4779 self.needed_locks[locking.LEVEL_NODE] = []
4780 self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
4782 self._migrater = TLMigrateInstance(self, self.op.instance_name,
4783 self.op.live, self.op.cleanup)
4784 self.tasklets = [self._migrater]
4786 def DeclareLocks(self, level):
4787 if level == locking.LEVEL_NODE:
4788 self._LockInstancesNodes()
4790 def BuildHooksEnv(self):
4793 This runs on master, primary and secondary nodes of the instance.
4796 instance = self._migrater.instance
4797 source_node = instance.primary_node
4798 target_node = instance.secondary_nodes[0]
4799 env = _BuildInstanceHookEnvByObject(self, instance)
4800 env["MIGRATE_LIVE"] = self.op.live
4801 env["MIGRATE_CLEANUP"] = self.op.cleanup
4803 "OLD_PRIMARY": source_node,
4804 "OLD_SECONDARY": target_node,
4805 "NEW_PRIMARY": target_node,
4806 "NEW_SECONDARY": source_node,
4808 nl = [self.cfg.GetMasterNode()] + list(instance.secondary_nodes)
4810 nl_post.append(source_node)
4811 return env, nl, nl_post
4814 class LUMoveInstance(LogicalUnit):
4815 """Move an instance by data-copying.
4818 HPATH = "instance-move"
4819 HTYPE = constants.HTYPE_INSTANCE
4820 _OP_REQP = ["instance_name", "target_node"]
4823 def CheckArguments(self):
4824 """Check the arguments.
4827 self.shutdown_timeout = getattr(self.op, "shutdown_timeout",
4828 constants.DEFAULT_SHUTDOWN_TIMEOUT)
4830 def ExpandNames(self):
4831 self._ExpandAndLockInstance()
4832 target_node = _ExpandNodeName(self.cfg, self.op.target_node)
4833 self.op.target_node = target_node
4834 self.needed_locks[locking.LEVEL_NODE] = [target_node]
4835 self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_APPEND
4837 def DeclareLocks(self, level):
4838 if level == locking.LEVEL_NODE:
4839 self._LockInstancesNodes(primary_only=True)
4841 def BuildHooksEnv(self):
4844 This runs on master, primary and secondary nodes of the instance.
4848 "TARGET_NODE": self.op.target_node,
4849 "SHUTDOWN_TIMEOUT": self.shutdown_timeout,
4851 env.update(_BuildInstanceHookEnvByObject(self, self.instance))
4852 nl = [self.cfg.GetMasterNode()] + [self.instance.primary_node,
4853 self.op.target_node]
4856 def CheckPrereq(self):
4857 """Check prerequisites.
4859 This checks that the instance is in the cluster.
4862 self.instance = instance = self.cfg.GetInstanceInfo(self.op.instance_name)
4863 assert self.instance is not None, \
4864 "Cannot retrieve locked instance %s" % self.op.instance_name
4866 node = self.cfg.GetNodeInfo(self.op.target_node)
4867 assert node is not None, \
4868 "Cannot retrieve locked node %s" % self.op.target_node
4870 self.target_node = target_node = node.name
4872 if target_node == instance.primary_node:
4873 raise errors.OpPrereqError("Instance %s is already on the node %s" %
4874 (instance.name, target_node),
4877 bep = self.cfg.GetClusterInfo().FillBE(instance)
4879 for idx, dsk in enumerate(instance.disks):
4880 if dsk.dev_type not in (constants.LD_LV, constants.LD_FILE):
4881 raise errors.OpPrereqError("Instance disk %d has a complex layout,"
4882 " cannot copy" % idx, errors.ECODE_STATE)
4884 _CheckNodeOnline(self, target_node)
4885 _CheckNodeNotDrained(self, target_node)
4887 if instance.admin_up:
4888 # check memory requirements on the secondary node
4889 _CheckNodeFreeMemory(self, target_node, "failing over instance %s" %
4890 instance.name, bep[constants.BE_MEMORY],
4891 instance.hypervisor)
4893 self.LogInfo("Not checking memory on the secondary node as"
4894 " instance will not be started")
4896 # check bridge existance
4897 _CheckInstanceBridgesExist(self, instance, node=target_node)
4899 def Exec(self, feedback_fn):
4900 """Move an instance.
4902 The move is done by shutting it down on its present node, copying
4903 the data over (slow) and starting it on the new node.
4906 instance = self.instance
4908 source_node = instance.primary_node
4909 target_node = self.target_node
4911 self.LogInfo("Shutting down instance %s on source node %s",
4912 instance.name, source_node)
4914 result = self.rpc.call_instance_shutdown(source_node, instance,
4915 self.shutdown_timeout)
4916 msg = result.fail_msg
4918 if self.op.ignore_consistency:
4919 self.proc.LogWarning("Could not shutdown instance %s on node %s."
4920 " Proceeding anyway. Please make sure node"
4921 " %s is down. Error details: %s",
4922 instance.name, source_node, source_node, msg)
4924 raise errors.OpExecError("Could not shutdown instance %s on"
4926 (instance.name, source_node, msg))
4928 # create the target disks
4930 _CreateDisks(self, instance, target_node=target_node)
4931 except errors.OpExecError:
4932 self.LogWarning("Device creation failed, reverting...")
4934 _RemoveDisks(self, instance, target_node=target_node)
4936 self.cfg.ReleaseDRBDMinors(instance.name)
4939 cluster_name = self.cfg.GetClusterInfo().cluster_name
4942 # activate, get path, copy the data over
4943 for idx, disk in enumerate(instance.disks):
4944 self.LogInfo("Copying data for disk %d", idx)
4945 result = self.rpc.call_blockdev_assemble(target_node, disk,
4946 instance.name, True)
4948 self.LogWarning("Can't assemble newly created disk %d: %s",
4949 idx, result.fail_msg)
4950 errs.append(result.fail_msg)
4952 dev_path = result.payload
4953 result = self.rpc.call_blockdev_export(source_node, disk,
4954 target_node, dev_path,
4957 self.LogWarning("Can't copy data over for disk %d: %s",
4958 idx, result.fail_msg)
4959 errs.append(result.fail_msg)
4963 self.LogWarning("Some disks failed to copy, aborting")
4965 _RemoveDisks(self, instance, target_node=target_node)
4967 self.cfg.ReleaseDRBDMinors(instance.name)
4968 raise errors.OpExecError("Errors during disk copy: %s" %
4971 instance.primary_node = target_node
4972 self.cfg.Update(instance, feedback_fn)
4974 self.LogInfo("Removing the disks on the original node")
4975 _RemoveDisks(self, instance, target_node=source_node)
4977 # Only start the instance if it's marked as up
4978 if instance.admin_up:
4979 self.LogInfo("Starting instance %s on node %s",
4980 instance.name, target_node)
4982 disks_ok, _ = _AssembleInstanceDisks(self, instance,
4983 ignore_secondaries=True)
4985 _ShutdownInstanceDisks(self, instance)
4986 raise errors.OpExecError("Can't activate the instance's disks")
4988 result = self.rpc.call_instance_start(target_node, instance, None, None)
4989 msg = result.fail_msg
4991 _ShutdownInstanceDisks(self, instance)
4992 raise errors.OpExecError("Could not start instance %s on node %s: %s" %
4993 (instance.name, target_node, msg))
4996 class LUMigrateNode(LogicalUnit):
4997 """Migrate all instances from a node.
5000 HPATH = "node-migrate"
5001 HTYPE = constants.HTYPE_NODE
5002 _OP_REQP = ["node_name", "live"]
5005 def ExpandNames(self):
5006 self.op.node_name = _ExpandNodeName(self.cfg, self.op.node_name)
5008 self.needed_locks = {
5009 locking.LEVEL_NODE: [self.op.node_name],
5012 self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_APPEND
5014 # Create tasklets for migrating instances for all instances on this node
5018 for inst in _GetNodePrimaryInstances(self.cfg, self.op.node_name):
5019 logging.debug("Migrating instance %s", inst.name)
5020 names.append(inst.name)
5022 tasklets.append(TLMigrateInstance(self, inst.name, self.op.live, False))
5024 self.tasklets = tasklets
5026 # Declare instance locks
5027 self.needed_locks[locking.LEVEL_INSTANCE] = names
5029 def DeclareLocks(self, level):
5030 if level == locking.LEVEL_NODE:
5031 self._LockInstancesNodes()
5033 def BuildHooksEnv(self):
5036 This runs on the master, the primary and all the secondaries.
5040 "NODE_NAME": self.op.node_name,
5043 nl = [self.cfg.GetMasterNode()]
5045 return (env, nl, nl)
5048 class TLMigrateInstance(Tasklet):
5049 def __init__(self, lu, instance_name, live, cleanup):
5050 """Initializes this class.
5053 Tasklet.__init__(self, lu)
5056 self.instance_name = instance_name
5058 self.cleanup = cleanup
5060 def CheckPrereq(self):
5061 """Check prerequisites.
5063 This checks that the instance is in the cluster.
5066 instance_name = _ExpandInstanceName(self.lu.cfg, self.instance_name)
5067 instance = self.cfg.GetInstanceInfo(instance_name)
5068 assert instance is not None
5070 if instance.disk_template != constants.DT_DRBD8:
5071 raise errors.OpPrereqError("Instance's disk layout is not"
5072 " drbd8, cannot migrate.", errors.ECODE_STATE)
5074 secondary_nodes = instance.secondary_nodes
5075 if not secondary_nodes:
5076 raise errors.ConfigurationError("No secondary node but using"
5077 " drbd8 disk template")
5079 i_be = self.cfg.GetClusterInfo().FillBE(instance)
5081 target_node = secondary_nodes[0]
5082 # check memory requirements on the secondary node
5083 _CheckNodeFreeMemory(self, target_node, "migrating instance %s" %
5084 instance.name, i_be[constants.BE_MEMORY],
5085 instance.hypervisor)
5087 # check bridge existance
5088 _CheckInstanceBridgesExist(self, instance, node=target_node)
5090 if not self.cleanup:
5091 _CheckNodeNotDrained(self, target_node)
5092 result = self.rpc.call_instance_migratable(instance.primary_node,
5094 result.Raise("Can't migrate, please use failover",
5095 prereq=True, ecode=errors.ECODE_STATE)
5097 self.instance = instance
5099 def _WaitUntilSync(self):
5100 """Poll with custom rpc for disk sync.
5102 This uses our own step-based rpc call.
5105 self.feedback_fn("* wait until resync is done")
5109 result = self.rpc.call_drbd_wait_sync(self.all_nodes,
5111 self.instance.disks)
5113 for node, nres in result.items():
5114 nres.Raise("Cannot resync disks on node %s" % node)
5115 node_done, node_percent = nres.payload
5116 all_done = all_done and node_done
5117 if node_percent is not None:
5118 min_percent = min(min_percent, node_percent)
5120 if min_percent < 100:
5121 self.feedback_fn(" - progress: %.1f%%" % min_percent)
5124 def _EnsureSecondary(self, node):
5125 """Demote a node to secondary.
5128 self.feedback_fn("* switching node %s to secondary mode" % node)
5130 for dev in self.instance.disks:
5131 self.cfg.SetDiskID(dev, node)
5133 result = self.rpc.call_blockdev_close(node, self.instance.name,
5134 self.instance.disks)
5135 result.Raise("Cannot change disk to secondary on node %s" % node)
5137 def _GoStandalone(self):
5138 """Disconnect from the network.
5141 self.feedback_fn("* changing into standalone mode")
5142 result = self.rpc.call_drbd_disconnect_net(self.all_nodes, self.nodes_ip,
5143 self.instance.disks)
5144 for node, nres in result.items():
5145 nres.Raise("Cannot disconnect disks node %s" % node)
5147 def _GoReconnect(self, multimaster):
5148 """Reconnect to the network.
5154 msg = "single-master"
5155 self.feedback_fn("* changing disks into %s mode" % msg)
5156 result = self.rpc.call_drbd_attach_net(self.all_nodes, self.nodes_ip,
5157 self.instance.disks,
5158 self.instance.name, multimaster)
5159 for node, nres in result.items():
5160 nres.Raise("Cannot change disks config on node %s" % node)
5162 def _ExecCleanup(self):
5163 """Try to cleanup after a failed migration.
5165 The cleanup is done by:
5166 - check that the instance is running only on one node
5167 (and update the config if needed)
5168 - change disks on its secondary node to secondary
5169 - wait until disks are fully synchronized
5170 - disconnect from the network
5171 - change disks into single-master mode
5172 - wait again until disks are fully synchronized
5175 instance = self.instance
5176 target_node = self.target_node
5177 source_node = self.source_node
5179 # check running on only one node
5180 self.feedback_fn("* checking where the instance actually runs"
5181 " (if this hangs, the hypervisor might be in"
5183 ins_l = self.rpc.call_instance_list(self.all_nodes, [instance.hypervisor])
5184 for node, result in ins_l.items():
5185 result.Raise("Can't contact node %s" % node)
5187 runningon_source = instance.name in ins_l[source_node].payload
5188 runningon_target = instance.name in ins_l[target_node].payload
5190 if runningon_source and runningon_target:
5191 raise errors.OpExecError("Instance seems to be running on two nodes,"
5192 " or the hypervisor is confused. You will have"
5193 " to ensure manually that it runs only on one"
5194 " and restart this operation.")
5196 if not (runningon_source or runningon_target):
5197 raise errors.OpExecError("Instance does not seem to be running at all."
5198 " In this case, it's safer to repair by"
5199 " running 'gnt-instance stop' to ensure disk"
5200 " shutdown, and then restarting it.")
5202 if runningon_target:
5203 # the migration has actually succeeded, we need to update the config
5204 self.feedback_fn("* instance running on secondary node (%s),"
5205 " updating config" % target_node)
5206 instance.primary_node = target_node
5207 self.cfg.Update(instance, self.feedback_fn)
5208 demoted_node = source_node
5210 self.feedback_fn("* instance confirmed to be running on its"
5211 " primary node (%s)" % source_node)
5212 demoted_node = target_node
5214 self._EnsureSecondary(demoted_node)
5216 self._WaitUntilSync()
5217 except errors.OpExecError:
5218 # we ignore here errors, since if the device is standalone, it
5219 # won't be able to sync
5221 self._GoStandalone()
5222 self._GoReconnect(False)
5223 self._WaitUntilSync()
5225 self.feedback_fn("* done")
5227 def _RevertDiskStatus(self):
5228 """Try to revert the disk status after a failed migration.
5231 target_node = self.target_node
5233 self._EnsureSecondary(target_node)
5234 self._GoStandalone()
5235 self._GoReconnect(False)
5236 self._WaitUntilSync()
5237 except errors.OpExecError, err:
5238 self.lu.LogWarning("Migration failed and I can't reconnect the"
5239 " drives: error '%s'\n"
5240 "Please look and recover the instance status" %
5243 def _AbortMigration(self):
5244 """Call the hypervisor code to abort a started migration.
5247 instance = self.instance
5248 target_node = self.target_node
5249 migration_info = self.migration_info
5251 abort_result = self.rpc.call_finalize_migration(target_node,
5255 abort_msg = abort_result.fail_msg
5257 logging.error("Aborting migration failed on target node %s: %s",
5258 target_node, abort_msg)
5259 # Don't raise an exception here, as we stil have to try to revert the
5260 # disk status, even if this step failed.
5262 def _ExecMigration(self):
5263 """Migrate an instance.
5265 The migrate is done by:
5266 - change the disks into dual-master mode
5267 - wait until disks are fully synchronized again
5268 - migrate the instance
5269 - change disks on the new secondary node (the old primary) to secondary
5270 - wait until disks are fully synchronized
5271 - change disks into single-master mode
5274 instance = self.instance
5275 target_node = self.target_node
5276 source_node = self.source_node
5278 self.feedback_fn("* checking disk consistency between source and target")
5279 for dev in instance.disks:
5280 if not _CheckDiskConsistency(self, dev, target_node, False):
5281 raise errors.OpExecError("Disk %s is degraded or not fully"
5282 " synchronized on target node,"
5283 " aborting migrate." % dev.iv_name)
5285 # First get the migration information from the remote node
5286 result = self.rpc.call_migration_info(source_node, instance)
5287 msg = result.fail_msg
5289 log_err = ("Failed fetching source migration information from %s: %s" %
5291 logging.error(log_err)
5292 raise errors.OpExecError(log_err)
5294 self.migration_info = migration_info = result.payload
5296 # Then switch the disks to master/master mode
5297 self._EnsureSecondary(target_node)
5298 self._GoStandalone()
5299 self._GoReconnect(True)
5300 self._WaitUntilSync()
5302 self.feedback_fn("* preparing %s to accept the instance" % target_node)
5303 result = self.rpc.call_accept_instance(target_node,
5306 self.nodes_ip[target_node])
5308 msg = result.fail_msg
5310 logging.error("Instance pre-migration failed, trying to revert"
5311 " disk status: %s", msg)
5312 self.feedback_fn("Pre-migration failed, aborting")
5313 self._AbortMigration()
5314 self._RevertDiskStatus()
5315 raise errors.OpExecError("Could not pre-migrate instance %s: %s" %
5316 (instance.name, msg))
5318 self.feedback_fn("* migrating instance to %s" % target_node)
5320 result = self.rpc.call_instance_migrate(source_node, instance,
5321 self.nodes_ip[target_node],
5323 msg = result.fail_msg
5325 logging.error("Instance migration failed, trying to revert"
5326 " disk status: %s", msg)
5327 self.feedback_fn("Migration failed, aborting")
5328 self._AbortMigration()
5329 self._RevertDiskStatus()
5330 raise errors.OpExecError("Could not migrate instance %s: %s" %
5331 (instance.name, msg))
5334 instance.primary_node = target_node
5335 # distribute new instance config to the other nodes
5336 self.cfg.Update(instance, self.feedback_fn)
5338 result = self.rpc.call_finalize_migration(target_node,
5342 msg = result.fail_msg
5344 logging.error("Instance migration succeeded, but finalization failed:"
5346 raise errors.OpExecError("Could not finalize instance migration: %s" %
5349 self._EnsureSecondary(source_node)
5350 self._WaitUntilSync()
5351 self._GoStandalone()
5352 self._GoReconnect(False)
5353 self._WaitUntilSync()
5355 self.feedback_fn("* done")
5357 def Exec(self, feedback_fn):
5358 """Perform the migration.
5361 feedback_fn("Migrating instance %s" % self.instance.name)
5363 self.feedback_fn = feedback_fn
5365 self.source_node = self.instance.primary_node
5366 self.target_node = self.instance.secondary_nodes[0]
5367 self.all_nodes = [self.source_node, self.target_node]
5369 self.source_node: self.cfg.GetNodeInfo(self.source_node).secondary_ip,
5370 self.target_node: self.cfg.GetNodeInfo(self.target_node).secondary_ip,
5374 return self._ExecCleanup()
5376 return self._ExecMigration()
5379 def _CreateBlockDev(lu, node, instance, device, force_create,
5381 """Create a tree of block devices on a given node.
5383 If this device type has to be created on secondaries, create it and
5386 If not, just recurse to children keeping the same 'force' value.
5388 @param lu: the lu on whose behalf we execute
5389 @param node: the node on which to create the device
5390 @type instance: L{objects.Instance}
5391 @param instance: the instance which owns the device
5392 @type device: L{objects.Disk}
5393 @param device: the device to create
5394 @type force_create: boolean
5395 @param force_create: whether to force creation of this device; this
5396 will be change to True whenever we find a device which has
5397 CreateOnSecondary() attribute
5398 @param info: the extra 'metadata' we should attach to the device
5399 (this will be represented as a LVM tag)
5400 @type force_open: boolean
5401 @param force_open: this parameter will be passes to the
5402 L{backend.BlockdevCreate} function where it specifies
5403 whether we run on primary or not, and it affects both
5404 the child assembly and the device own Open() execution
5407 if device.CreateOnSecondary():
5411 for child in device.children:
5412 _CreateBlockDev(lu, node, instance, child, force_create,
5415 if not force_create:
5418 _CreateSingleBlockDev(lu, node, instance, device, info, force_open)
5421 def _CreateSingleBlockDev(lu, node, instance, device, info, force_open):
5422 """Create a single block device on a given node.
5424 This will not recurse over children of the device, so they must be
5427 @param lu: the lu on whose behalf we execute
5428 @param node: the node on which to create the device
5429 @type instance: L{objects.Instance}
5430 @param instance: the instance which owns the device
5431 @type device: L{objects.Disk}
5432 @param device: the device to create
5433 @param info: the extra 'metadata' we should attach to the device
5434 (this will be represented as a LVM tag)
5435 @type force_open: boolean
5436 @param force_open: this parameter will be passes to the
5437 L{backend.BlockdevCreate} function where it specifies
5438 whether we run on primary or not, and it affects both
5439 the child assembly and the device own Open() execution
5442 lu.cfg.SetDiskID(device, node)
5443 result = lu.rpc.call_blockdev_create(node, device, device.size,
5444 instance.name, force_open, info)
5445 result.Raise("Can't create block device %s on"
5446 " node %s for instance %s" % (device, node, instance.name))
5447 if device.physical_id is None:
5448 device.physical_id = result.payload
5451 def _GenerateUniqueNames(lu, exts):
5452 """Generate a suitable LV name.
5454 This will generate a logical volume name for the given instance.
5459 new_id = lu.cfg.GenerateUniqueID(lu.proc.GetECId())
5460 results.append("%s%s" % (new_id, val))
5464 def _GenerateDRBD8Branch(lu, primary, secondary, size, names, iv_name,
5466 """Generate a drbd8 device complete with its children.
5469 port = lu.cfg.AllocatePort()
5470 vgname = lu.cfg.GetVGName()
5471 shared_secret = lu.cfg.GenerateDRBDSecret(lu.proc.GetECId())
5472 dev_data = objects.Disk(dev_type=constants.LD_LV, size=size,
5473 logical_id=(vgname, names[0]))
5474 dev_meta = objects.Disk(dev_type=constants.LD_LV, size=128,
5475 logical_id=(vgname, names[1]))
5476 drbd_dev = objects.Disk(dev_type=constants.LD_DRBD8, size=size,
5477 logical_id=(primary, secondary, port,
5480 children=[dev_data, dev_meta],
5485 def _GenerateDiskTemplate(lu, template_name,
5486 instance_name, primary_node,
5487 secondary_nodes, disk_info,
5488 file_storage_dir, file_driver,
5490 """Generate the entire disk layout for a given template type.
5493 #TODO: compute space requirements
5495 vgname = lu.cfg.GetVGName()
5496 disk_count = len(disk_info)
5498 if template_name == constants.DT_DISKLESS:
5500 elif template_name == constants.DT_PLAIN:
5501 if len(secondary_nodes) != 0:
5502 raise errors.ProgrammerError("Wrong template configuration")
5504 names = _GenerateUniqueNames(lu, [".disk%d" % (base_index + i)
5505 for i in range(disk_count)])
5506 for idx, disk in enumerate(disk_info):
5507 disk_index = idx + base_index
5508 disk_dev = objects.Disk(dev_type=constants.LD_LV, size=disk["size"],
5509 logical_id=(vgname, names[idx]),
5510 iv_name="disk/%d" % disk_index,
5512 disks.append(disk_dev)
5513 elif template_name == constants.DT_DRBD8:
5514 if len(secondary_nodes) != 1:
5515 raise errors.ProgrammerError("Wrong template configuration")
5516 remote_node = secondary_nodes[0]
5517 minors = lu.cfg.AllocateDRBDMinor(
5518 [primary_node, remote_node] * len(disk_info), instance_name)
5521 for lv_prefix in _GenerateUniqueNames(lu, [".disk%d" % (base_index + i)
5522 for i in range(disk_count)]):
5523 names.append(lv_prefix + "_data")
5524 names.append(lv_prefix + "_meta")
5525 for idx, disk in enumerate(disk_info):
5526 disk_index = idx + base_index
5527 disk_dev = _GenerateDRBD8Branch(lu, primary_node, remote_node,
5528 disk["size"], names[idx*2:idx*2+2],
5529 "disk/%d" % disk_index,
5530 minors[idx*2], minors[idx*2+1])
5531 disk_dev.mode = disk["mode"]
5532 disks.append(disk_dev)
5533 elif template_name == constants.DT_FILE:
5534 if len(secondary_nodes) != 0:
5535 raise errors.ProgrammerError("Wrong template configuration")
5537 for idx, disk in enumerate(disk_info):
5538 disk_index = idx + base_index
5539 disk_dev = objects.Disk(dev_type=constants.LD_FILE, size=disk["size"],
5540 iv_name="disk/%d" % disk_index,
5541 logical_id=(file_driver,
5542 "%s/disk%d" % (file_storage_dir,
5545 disks.append(disk_dev)
5547 raise errors.ProgrammerError("Invalid disk template '%s'" % template_name)
5551 def _GetInstanceInfoText(instance):
5552 """Compute that text that should be added to the disk's metadata.
5555 return "originstname+%s" % instance.name
5558 def _CreateDisks(lu, instance, to_skip=None, target_node=None):
5559 """Create all disks for an instance.
5561 This abstracts away some work from AddInstance.
5563 @type lu: L{LogicalUnit}
5564 @param lu: the logical unit on whose behalf we execute
5565 @type instance: L{objects.Instance}
5566 @param instance: the instance whose disks we should create
5568 @param to_skip: list of indices to skip
5569 @type target_node: string
5570 @param target_node: if passed, overrides the target node for creation
5572 @return: the success of the creation
5575 info = _GetInstanceInfoText(instance)
5576 if target_node is None:
5577 pnode = instance.primary_node
5578 all_nodes = instance.all_nodes
5583 if instance.disk_template == constants.DT_FILE:
5584 file_storage_dir = os.path.dirname(instance.disks[0].logical_id[1])
5585 result = lu.rpc.call_file_storage_dir_create(pnode, file_storage_dir)
5587 result.Raise("Failed to create directory '%s' on"
5588 " node %s" % (file_storage_dir, pnode))
5590 # Note: this needs to be kept in sync with adding of disks in
5591 # LUSetInstanceParams
5592 for idx, device in enumerate(instance.disks):
5593 if to_skip and idx in to_skip:
5595 logging.info("Creating volume %s for instance %s",
5596 device.iv_name, instance.name)
5598 for node in all_nodes:
5599 f_create = node == pnode
5600 _CreateBlockDev(lu, node, instance, device, f_create, info, f_create)
5603 def _RemoveDisks(lu, instance, target_node=None):
5604 """Remove all disks for an instance.
5606 This abstracts away some work from `AddInstance()` and
5607 `RemoveInstance()`. Note that in case some of the devices couldn't
5608 be removed, the removal will continue with the other ones (compare
5609 with `_CreateDisks()`).
5611 @type lu: L{LogicalUnit}
5612 @param lu: the logical unit on whose behalf we execute
5613 @type instance: L{objects.Instance}
5614 @param instance: the instance whose disks we should remove
5615 @type target_node: string
5616 @param target_node: used to override the node on which to remove the disks
5618 @return: the success of the removal
5621 logging.info("Removing block devices for instance %s", instance.name)
5624 for device in instance.disks:
5626 edata = [(target_node, device)]
5628 edata = device.ComputeNodeTree(instance.primary_node)
5629 for node, disk in edata:
5630 lu.cfg.SetDiskID(disk, node)
5631 msg = lu.rpc.call_blockdev_remove(node, disk).fail_msg
5633 lu.LogWarning("Could not remove block device %s on node %s,"
5634 " continuing anyway: %s", device.iv_name, node, msg)
5637 if instance.disk_template == constants.DT_FILE:
5638 file_storage_dir = os.path.dirname(instance.disks[0].logical_id[1])
5642 tgt = instance.primary_node
5643 result = lu.rpc.call_file_storage_dir_remove(tgt, file_storage_dir)
5645 lu.LogWarning("Could not remove directory '%s' on node %s: %s",
5646 file_storage_dir, instance.primary_node, result.fail_msg)
5652 def _ComputeDiskSize(disk_template, disks):
5653 """Compute disk size requirements in the volume group
5656 # Required free disk space as a function of disk and swap space
5658 constants.DT_DISKLESS: None,
5659 constants.DT_PLAIN: sum(d["size"] for d in disks),
5660 # 128 MB are added for drbd metadata for each disk
5661 constants.DT_DRBD8: sum(d["size"] + 128 for d in disks),
5662 constants.DT_FILE: None,
5665 if disk_template not in req_size_dict:
5666 raise errors.ProgrammerError("Disk template '%s' size requirement"
5667 " is unknown" % disk_template)
5669 return req_size_dict[disk_template]
5672 def _CheckHVParams(lu, nodenames, hvname, hvparams):
5673 """Hypervisor parameter validation.
5675 This function abstract the hypervisor parameter validation to be
5676 used in both instance create and instance modify.
5678 @type lu: L{LogicalUnit}
5679 @param lu: the logical unit for which we check
5680 @type nodenames: list
5681 @param nodenames: the list of nodes on which we should check
5682 @type hvname: string
5683 @param hvname: the name of the hypervisor we should use
5684 @type hvparams: dict
5685 @param hvparams: the parameters which we need to check
5686 @raise errors.OpPrereqError: if the parameters are not valid
5689 hvinfo = lu.rpc.call_hypervisor_validate_params(nodenames,
5692 for node in nodenames:
5696 info.Raise("Hypervisor parameter validation failed on node %s" % node)
5699 class LUCreateInstance(LogicalUnit):
5700 """Create an instance.
5703 HPATH = "instance-add"
5704 HTYPE = constants.HTYPE_INSTANCE
5705 _OP_REQP = ["instance_name", "disks", "disk_template",
5707 "wait_for_sync", "ip_check", "nics",
5708 "hvparams", "beparams"]
5711 def CheckArguments(self):
5715 # do not require name_check to ease forward/backward compatibility
5717 if not hasattr(self.op, "name_check"):
5718 self.op.name_check = True
5719 # validate/normalize the instance name
5720 self.op.instance_name = utils.HostInfo.NormalizeName(self.op.instance_name)
5721 if self.op.ip_check and not self.op.name_check:
5722 # TODO: make the ip check more flexible and not depend on the name check
5723 raise errors.OpPrereqError("Cannot do ip checks without a name check",
5725 if (self.op.disk_template == constants.DT_FILE and
5726 not constants.ENABLE_FILE_STORAGE):
5727 raise errors.OpPrereqError("File storage disabled at configure time",
5730 def ExpandNames(self):
5731 """ExpandNames for CreateInstance.
5733 Figure out the right locks for instance creation.
5736 self.needed_locks = {}
5738 # set optional parameters to none if they don't exist
5739 for attr in ["pnode", "snode", "iallocator", "hypervisor"]:
5740 if not hasattr(self.op, attr):
5741 setattr(self.op, attr, None)
5743 # cheap checks, mostly valid constants given
5745 # verify creation mode
5746 if self.op.mode not in (constants.INSTANCE_CREATE,
5747 constants.INSTANCE_IMPORT):
5748 raise errors.OpPrereqError("Invalid instance creation mode '%s'" %
5749 self.op.mode, errors.ECODE_INVAL)
5751 # disk template and mirror node verification
5752 if self.op.disk_template not in constants.DISK_TEMPLATES:
5753 raise errors.OpPrereqError("Invalid disk template name",
5756 if self.op.hypervisor is None:
5757 self.op.hypervisor = self.cfg.GetHypervisorType()
5759 cluster = self.cfg.GetClusterInfo()
5760 enabled_hvs = cluster.enabled_hypervisors
5761 if self.op.hypervisor not in enabled_hvs:
5762 raise errors.OpPrereqError("Selected hypervisor (%s) not enabled in the"
5763 " cluster (%s)" % (self.op.hypervisor,
5764 ",".join(enabled_hvs)),
5767 # check hypervisor parameter syntax (locally)
5768 utils.ForceDictType(self.op.hvparams, constants.HVS_PARAMETER_TYPES)
5769 filled_hvp = objects.FillDict(cluster.hvparams[self.op.hypervisor],
5771 hv_type = hypervisor.GetHypervisor(self.op.hypervisor)
5772 hv_type.CheckParameterSyntax(filled_hvp)
5773 self.hv_full = filled_hvp
5774 # check that we don't specify global parameters on an instance
5775 _CheckGlobalHvParams(self.op.hvparams)
5777 # fill and remember the beparams dict
5778 utils.ForceDictType(self.op.beparams, constants.BES_PARAMETER_TYPES)
5779 self.be_full = objects.FillDict(cluster.beparams[constants.PP_DEFAULT],
5782 #### instance parameters check
5784 # instance name verification
5785 if self.op.name_check:
5786 hostname1 = utils.GetHostInfo(self.op.instance_name)
5787 self.op.instance_name = instance_name = hostname1.name
5788 # used in CheckPrereq for ip ping check
5789 self.check_ip = hostname1.ip
5791 instance_name = self.op.instance_name
5792 self.check_ip = None
5794 # this is just a preventive check, but someone might still add this
5795 # instance in the meantime, and creation will fail at lock-add time
5796 if instance_name in self.cfg.GetInstanceList():
5797 raise errors.OpPrereqError("Instance '%s' is already in the cluster" %
5798 instance_name, errors.ECODE_EXISTS)
5800 self.add_locks[locking.LEVEL_INSTANCE] = instance_name
5804 for idx, nic in enumerate(self.op.nics):
5805 nic_mode_req = nic.get("mode", None)
5806 nic_mode = nic_mode_req
5807 if nic_mode is None:
5808 nic_mode = cluster.nicparams[constants.PP_DEFAULT][constants.NIC_MODE]
5810 # in routed mode, for the first nic, the default ip is 'auto'
5811 if nic_mode == constants.NIC_MODE_ROUTED and idx == 0:
5812 default_ip_mode = constants.VALUE_AUTO
5814 default_ip_mode = constants.VALUE_NONE
5816 # ip validity checks
5817 ip = nic.get("ip", default_ip_mode)
5818 if ip is None or ip.lower() == constants.VALUE_NONE:
5820 elif ip.lower() == constants.VALUE_AUTO:
5821 if not self.op.name_check:
5822 raise errors.OpPrereqError("IP address set to auto but name checks"
5823 " have been skipped. Aborting.",
5825 nic_ip = hostname1.ip
5827 if not utils.IsValidIP(ip):
5828 raise errors.OpPrereqError("Given IP address '%s' doesn't look"
5829 " like a valid IP" % ip,
5833 # TODO: check the ip address for uniqueness
5834 if nic_mode == constants.NIC_MODE_ROUTED and not nic_ip:
5835 raise errors.OpPrereqError("Routed nic mode requires an ip address",
5838 # MAC address verification
5839 mac = nic.get("mac", constants.VALUE_AUTO)
5840 if mac not in (constants.VALUE_AUTO, constants.VALUE_GENERATE):
5841 mac = utils.NormalizeAndValidateMac(mac)
5844 self.cfg.ReserveMAC(mac, self.proc.GetECId())
5845 except errors.ReservationError:
5846 raise errors.OpPrereqError("MAC address %s already in use"
5847 " in cluster" % mac,
5848 errors.ECODE_NOTUNIQUE)
5850 # bridge verification
5851 bridge = nic.get("bridge", None)
5852 link = nic.get("link", None)
5854 raise errors.OpPrereqError("Cannot pass 'bridge' and 'link'"
5855 " at the same time", errors.ECODE_INVAL)
5856 elif bridge and nic_mode == constants.NIC_MODE_ROUTED:
5857 raise errors.OpPrereqError("Cannot pass 'bridge' on a routed nic",
5864 nicparams[constants.NIC_MODE] = nic_mode_req
5866 nicparams[constants.NIC_LINK] = link
5868 check_params = objects.FillDict(cluster.nicparams[constants.PP_DEFAULT],
5870 objects.NIC.CheckParameterSyntax(check_params)
5871 self.nics.append(objects.NIC(mac=mac, ip=nic_ip, nicparams=nicparams))
5873 # disk checks/pre-build
5875 for disk in self.op.disks:
5876 mode = disk.get("mode", constants.DISK_RDWR)
5877 if mode not in constants.DISK_ACCESS_SET:
5878 raise errors.OpPrereqError("Invalid disk access mode '%s'" %
5879 mode, errors.ECODE_INVAL)
5880 size = disk.get("size", None)
5882 raise errors.OpPrereqError("Missing disk size", errors.ECODE_INVAL)
5885 except (TypeError, ValueError):
5886 raise errors.OpPrereqError("Invalid disk size '%s'" % size,
5888 self.disks.append({"size": size, "mode": mode})
5890 # file storage checks
5891 if (self.op.file_driver and
5892 not self.op.file_driver in constants.FILE_DRIVER):
5893 raise errors.OpPrereqError("Invalid file driver name '%s'" %
5894 self.op.file_driver, errors.ECODE_INVAL)
5896 if self.op.file_storage_dir and os.path.isabs(self.op.file_storage_dir):
5897 raise errors.OpPrereqError("File storage directory path not absolute",
5900 ### Node/iallocator related checks
5901 if [self.op.iallocator, self.op.pnode].count(None) != 1:
5902 raise errors.OpPrereqError("One and only one of iallocator and primary"
5903 " node must be given",
5906 if self.op.iallocator:
5907 self.needed_locks[locking.LEVEL_NODE] = locking.ALL_SET
5909 self.op.pnode = _ExpandNodeName(self.cfg, self.op.pnode)
5910 nodelist = [self.op.pnode]
5911 if self.op.snode is not None:
5912 self.op.snode = _ExpandNodeName(self.cfg, self.op.snode)
5913 nodelist.append(self.op.snode)
5914 self.needed_locks[locking.LEVEL_NODE] = nodelist
5916 # in case of import lock the source node too
5917 if self.op.mode == constants.INSTANCE_IMPORT:
5918 src_node = getattr(self.op, "src_node", None)
5919 src_path = getattr(self.op, "src_path", None)
5921 if src_path is None:
5922 self.op.src_path = src_path = self.op.instance_name
5924 if src_node is None:
5925 self.needed_locks[locking.LEVEL_NODE] = locking.ALL_SET
5926 self.op.src_node = None
5927 if os.path.isabs(src_path):
5928 raise errors.OpPrereqError("Importing an instance from an absolute"
5929 " path requires a source node option.",
5932 self.op.src_node = src_node = _ExpandNodeName(self.cfg, src_node)
5933 if self.needed_locks[locking.LEVEL_NODE] is not locking.ALL_SET:
5934 self.needed_locks[locking.LEVEL_NODE].append(src_node)
5935 if not os.path.isabs(src_path):
5936 self.op.src_path = src_path = \
5937 utils.PathJoin(constants.EXPORT_DIR, src_path)
5939 # On import force_variant must be True, because if we forced it at
5940 # initial install, our only chance when importing it back is that it
5942 self.op.force_variant = True
5944 else: # INSTANCE_CREATE
5945 if getattr(self.op, "os_type", None) is None:
5946 raise errors.OpPrereqError("No guest OS specified",
5948 self.op.force_variant = getattr(self.op, "force_variant", False)
5950 def _RunAllocator(self):
5951 """Run the allocator based on input opcode.
5954 nics = [n.ToDict() for n in self.nics]
5955 ial = IAllocator(self.cfg, self.rpc,
5956 mode=constants.IALLOCATOR_MODE_ALLOC,
5957 name=self.op.instance_name,
5958 disk_template=self.op.disk_template,
5961 vcpus=self.be_full[constants.BE_VCPUS],
5962 mem_size=self.be_full[constants.BE_MEMORY],
5965 hypervisor=self.op.hypervisor,
5968 ial.Run(self.op.iallocator)
5971 raise errors.OpPrereqError("Can't compute nodes using"
5972 " iallocator '%s': %s" %
5973 (self.op.iallocator, ial.info),
5975 if len(ial.result) != ial.required_nodes:
5976 raise errors.OpPrereqError("iallocator '%s' returned invalid number"
5977 " of nodes (%s), required %s" %
5978 (self.op.iallocator, len(ial.result),
5979 ial.required_nodes), errors.ECODE_FAULT)
5980 self.op.pnode = ial.result[0]
5981 self.LogInfo("Selected nodes for instance %s via iallocator %s: %s",
5982 self.op.instance_name, self.op.iallocator,
5983 utils.CommaJoin(ial.result))
5984 if ial.required_nodes == 2:
5985 self.op.snode = ial.result[1]
5987 def BuildHooksEnv(self):
5990 This runs on master, primary and secondary nodes of the instance.
5994 "ADD_MODE": self.op.mode,
5996 if self.op.mode == constants.INSTANCE_IMPORT:
5997 env["SRC_NODE"] = self.op.src_node
5998 env["SRC_PATH"] = self.op.src_path
5999 env["SRC_IMAGES"] = self.src_images
6001 env.update(_BuildInstanceHookEnv(
6002 name=self.op.instance_name,
6003 primary_node=self.op.pnode,
6004 secondary_nodes=self.secondaries,
6005 status=self.op.start,
6006 os_type=self.op.os_type,
6007 memory=self.be_full[constants.BE_MEMORY],
6008 vcpus=self.be_full[constants.BE_VCPUS],
6009 nics=_NICListToTuple(self, self.nics),
6010 disk_template=self.op.disk_template,
6011 disks=[(d["size"], d["mode"]) for d in self.disks],
6014 hypervisor_name=self.op.hypervisor,
6017 nl = ([self.cfg.GetMasterNode(), self.op.pnode] +
6022 def CheckPrereq(self):
6023 """Check prerequisites.
6026 if (not self.cfg.GetVGName() and
6027 self.op.disk_template not in constants.DTS_NOT_LVM):
6028 raise errors.OpPrereqError("Cluster does not support lvm-based"
6029 " instances", errors.ECODE_STATE)
6031 if self.op.mode == constants.INSTANCE_IMPORT:
6032 src_node = self.op.src_node
6033 src_path = self.op.src_path
6035 if src_node is None:
6036 locked_nodes = self.acquired_locks[locking.LEVEL_NODE]
6037 exp_list = self.rpc.call_export_list(locked_nodes)
6039 for node in exp_list:
6040 if exp_list[node].fail_msg:
6042 if src_path in exp_list[node].payload:
6044 self.op.src_node = src_node = node
6045 self.op.src_path = src_path = utils.PathJoin(constants.EXPORT_DIR,
6049 raise errors.OpPrereqError("No export found for relative path %s" %
6050 src_path, errors.ECODE_INVAL)
6052 _CheckNodeOnline(self, src_node)
6053 result = self.rpc.call_export_info(src_node, src_path)
6054 result.Raise("No export or invalid export found in dir %s" % src_path)
6056 export_info = objects.SerializableConfigParser.Loads(str(result.payload))
6057 if not export_info.has_section(constants.INISECT_EXP):
6058 raise errors.ProgrammerError("Corrupted export config",
6059 errors.ECODE_ENVIRON)
6061 ei_version = export_info.get(constants.INISECT_EXP, 'version')
6062 if (int(ei_version) != constants.EXPORT_VERSION):
6063 raise errors.OpPrereqError("Wrong export version %s (wanted %d)" %
6064 (ei_version, constants.EXPORT_VERSION),
6065 errors.ECODE_ENVIRON)
6067 # Check that the new instance doesn't have less disks than the export
6068 instance_disks = len(self.disks)
6069 export_disks = export_info.getint(constants.INISECT_INS, 'disk_count')
6070 if instance_disks < export_disks:
6071 raise errors.OpPrereqError("Not enough disks to import."
6072 " (instance: %d, export: %d)" %
6073 (instance_disks, export_disks),
6076 self.op.os_type = export_info.get(constants.INISECT_EXP, 'os')
6078 for idx in range(export_disks):
6079 option = 'disk%d_dump' % idx
6080 if export_info.has_option(constants.INISECT_INS, option):
6081 # FIXME: are the old os-es, disk sizes, etc. useful?
6082 export_name = export_info.get(constants.INISECT_INS, option)
6083 image = utils.PathJoin(src_path, export_name)
6084 disk_images.append(image)
6086 disk_images.append(False)
6088 self.src_images = disk_images
6090 old_name = export_info.get(constants.INISECT_INS, 'name')
6091 # FIXME: int() here could throw a ValueError on broken exports
6092 exp_nic_count = int(export_info.get(constants.INISECT_INS, 'nic_count'))
6093 if self.op.instance_name == old_name:
6094 for idx, nic in enumerate(self.nics):
6095 if nic.mac == constants.VALUE_AUTO and exp_nic_count >= idx:
6096 nic_mac_ini = 'nic%d_mac' % idx
6097 nic.mac = export_info.get(constants.INISECT_INS, nic_mac_ini)
6099 # ENDIF: self.op.mode == constants.INSTANCE_IMPORT
6101 # ip ping checks (we use the same ip that was resolved in ExpandNames)
6102 if self.op.ip_check:
6103 if utils.TcpPing(self.check_ip, constants.DEFAULT_NODED_PORT):
6104 raise errors.OpPrereqError("IP %s of instance %s already in use" %
6105 (self.check_ip, self.op.instance_name),
6106 errors.ECODE_NOTUNIQUE)
6108 #### mac address generation
6109 # By generating here the mac address both the allocator and the hooks get
6110 # the real final mac address rather than the 'auto' or 'generate' value.
6111 # There is a race condition between the generation and the instance object
6112 # creation, which means that we know the mac is valid now, but we're not
6113 # sure it will be when we actually add the instance. If things go bad
6114 # adding the instance will abort because of a duplicate mac, and the
6115 # creation job will fail.
6116 for nic in self.nics:
6117 if nic.mac in (constants.VALUE_AUTO, constants.VALUE_GENERATE):
6118 nic.mac = self.cfg.GenerateMAC(self.proc.GetECId())
6122 if self.op.iallocator is not None:
6123 self._RunAllocator()
6125 #### node related checks
6127 # check primary node
6128 self.pnode = pnode = self.cfg.GetNodeInfo(self.op.pnode)
6129 assert self.pnode is not None, \
6130 "Cannot retrieve locked node %s" % self.op.pnode
6132 raise errors.OpPrereqError("Cannot use offline primary node '%s'" %
6133 pnode.name, errors.ECODE_STATE)
6135 raise errors.OpPrereqError("Cannot use drained primary node '%s'" %
6136 pnode.name, errors.ECODE_STATE)
6138 self.secondaries = []
6140 # mirror node verification
6141 if self.op.disk_template in constants.DTS_NET_MIRROR:
6142 if self.op.snode is None:
6143 raise errors.OpPrereqError("The networked disk templates need"
6144 " a mirror node", errors.ECODE_INVAL)
6145 if self.op.snode == pnode.name:
6146 raise errors.OpPrereqError("The secondary node cannot be the"
6147 " primary node.", errors.ECODE_INVAL)
6148 _CheckNodeOnline(self, self.op.snode)
6149 _CheckNodeNotDrained(self, self.op.snode)
6150 self.secondaries.append(self.op.snode)
6152 nodenames = [pnode.name] + self.secondaries
6154 req_size = _ComputeDiskSize(self.op.disk_template,
6157 # Check lv size requirements
6158 if req_size is not None:
6159 nodeinfo = self.rpc.call_node_info(nodenames, self.cfg.GetVGName(),
6161 for node in nodenames:
6162 info = nodeinfo[node]
6163 info.Raise("Cannot get current information from node %s" % node)
6165 vg_free = info.get('vg_free', None)
6166 if not isinstance(vg_free, int):
6167 raise errors.OpPrereqError("Can't compute free disk space on"
6168 " node %s" % node, errors.ECODE_ENVIRON)
6169 if req_size > vg_free:
6170 raise errors.OpPrereqError("Not enough disk space on target node %s."
6171 " %d MB available, %d MB required" %
6172 (node, vg_free, req_size),
6175 _CheckHVParams(self, nodenames, self.op.hypervisor, self.op.hvparams)
6178 result = self.rpc.call_os_get(pnode.name, self.op.os_type)
6179 result.Raise("OS '%s' not in supported os list for primary node %s" %
6180 (self.op.os_type, pnode.name),
6181 prereq=True, ecode=errors.ECODE_INVAL)
6182 if not self.op.force_variant:
6183 _CheckOSVariant(result.payload, self.op.os_type)
6185 _CheckNicsBridgesExist(self, self.nics, self.pnode.name)
6187 # memory check on primary node
6189 _CheckNodeFreeMemory(self, self.pnode.name,
6190 "creating instance %s" % self.op.instance_name,
6191 self.be_full[constants.BE_MEMORY],
6194 self.dry_run_result = list(nodenames)
6196 def Exec(self, feedback_fn):
6197 """Create and add the instance to the cluster.
6200 instance = self.op.instance_name
6201 pnode_name = self.pnode.name
6203 ht_kind = self.op.hypervisor
6204 if ht_kind in constants.HTS_REQ_PORT:
6205 network_port = self.cfg.AllocatePort()
6209 ##if self.op.vnc_bind_address is None:
6210 ## self.op.vnc_bind_address = constants.VNC_DEFAULT_BIND_ADDRESS
6212 # this is needed because os.path.join does not accept None arguments
6213 if self.op.file_storage_dir is None:
6214 string_file_storage_dir = ""
6216 string_file_storage_dir = self.op.file_storage_dir
6218 # build the full file storage dir path
6219 file_storage_dir = utils.PathJoin(self.cfg.GetFileStorageDir(),
6220 string_file_storage_dir, instance)
6223 disks = _GenerateDiskTemplate(self,
6224 self.op.disk_template,
6225 instance, pnode_name,
6229 self.op.file_driver,
6232 iobj = objects.Instance(name=instance, os=self.op.os_type,
6233 primary_node=pnode_name,
6234 nics=self.nics, disks=disks,
6235 disk_template=self.op.disk_template,
6237 network_port=network_port,
6238 beparams=self.op.beparams,
6239 hvparams=self.op.hvparams,
6240 hypervisor=self.op.hypervisor,
6243 feedback_fn("* creating instance disks...")
6245 _CreateDisks(self, iobj)
6246 except errors.OpExecError:
6247 self.LogWarning("Device creation failed, reverting...")
6249 _RemoveDisks(self, iobj)
6251 self.cfg.ReleaseDRBDMinors(instance)
6254 feedback_fn("adding instance %s to cluster config" % instance)
6256 self.cfg.AddInstance(iobj, self.proc.GetECId())
6258 # Declare that we don't want to remove the instance lock anymore, as we've
6259 # added the instance to the config
6260 del self.remove_locks[locking.LEVEL_INSTANCE]
6261 # Unlock all the nodes
6262 if self.op.mode == constants.INSTANCE_IMPORT:
6263 nodes_keep = [self.op.src_node]
6264 nodes_release = [node for node in self.acquired_locks[locking.LEVEL_NODE]
6265 if node != self.op.src_node]
6266 self.context.glm.release(locking.LEVEL_NODE, nodes_release)
6267 self.acquired_locks[locking.LEVEL_NODE] = nodes_keep
6269 self.context.glm.release(locking.LEVEL_NODE)
6270 del self.acquired_locks[locking.LEVEL_NODE]
6272 if self.op.wait_for_sync:
6273 disk_abort = not _WaitForSync(self, iobj)
6274 elif iobj.disk_template in constants.DTS_NET_MIRROR:
6275 # make sure the disks are not degraded (still sync-ing is ok)
6277 feedback_fn("* checking mirrors status")
6278 disk_abort = not _WaitForSync(self, iobj, oneshot=True)
6283 _RemoveDisks(self, iobj)
6284 self.cfg.RemoveInstance(iobj.name)
6285 # Make sure the instance lock gets removed
6286 self.remove_locks[locking.LEVEL_INSTANCE] = iobj.name
6287 raise errors.OpExecError("There are some degraded disks for"
6290 feedback_fn("creating os for instance %s on node %s" %
6291 (instance, pnode_name))
6293 if iobj.disk_template != constants.DT_DISKLESS:
6294 if self.op.mode == constants.INSTANCE_CREATE:
6295 feedback_fn("* running the instance OS create scripts...")
6296 # FIXME: pass debug option from opcode to backend
6297 result = self.rpc.call_instance_os_add(pnode_name, iobj, False,
6298 self.op.debug_level)
6299 result.Raise("Could not add os for instance %s"
6300 " on node %s" % (instance, pnode_name))
6302 elif self.op.mode == constants.INSTANCE_IMPORT:
6303 feedback_fn("* running the instance OS import scripts...")
6304 src_node = self.op.src_node
6305 src_images = self.src_images
6306 cluster_name = self.cfg.GetClusterName()
6307 # FIXME: pass debug option from opcode to backend
6308 import_result = self.rpc.call_instance_os_import(pnode_name, iobj,
6309 src_node, src_images,
6311 self.op.debug_level)
6312 msg = import_result.fail_msg
6314 self.LogWarning("Error while importing the disk images for instance"
6315 " %s on node %s: %s" % (instance, pnode_name, msg))
6317 # also checked in the prereq part
6318 raise errors.ProgrammerError("Unknown OS initialization mode '%s'"
6322 iobj.admin_up = True
6323 self.cfg.Update(iobj, feedback_fn)
6324 logging.info("Starting instance %s on node %s", instance, pnode_name)
6325 feedback_fn("* starting instance...")
6326 result = self.rpc.call_instance_start(pnode_name, iobj, None, None)
6327 result.Raise("Could not start instance")
6329 return list(iobj.all_nodes)
6332 class LUConnectConsole(NoHooksLU):
6333 """Connect to an instance's console.
6335 This is somewhat special in that it returns the command line that
6336 you need to run on the master node in order to connect to the
6340 _OP_REQP = ["instance_name"]
6343 def ExpandNames(self):
6344 self._ExpandAndLockInstance()
6346 def CheckPrereq(self):
6347 """Check prerequisites.
6349 This checks that the instance is in the cluster.
6352 self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
6353 assert self.instance is not None, \
6354 "Cannot retrieve locked instance %s" % self.op.instance_name
6355 _CheckNodeOnline(self, self.instance.primary_node)
6357 def Exec(self, feedback_fn):
6358 """Connect to the console of an instance
6361 instance = self.instance
6362 node = instance.primary_node
6364 node_insts = self.rpc.call_instance_list([node],
6365 [instance.hypervisor])[node]
6366 node_insts.Raise("Can't get node information from %s" % node)
6368 if instance.name not in node_insts.payload:
6369 raise errors.OpExecError("Instance %s is not running." % instance.name)
6371 logging.debug("Connecting to console of %s on %s", instance.name, node)
6373 hyper = hypervisor.GetHypervisor(instance.hypervisor)
6374 cluster = self.cfg.GetClusterInfo()
6375 # beparams and hvparams are passed separately, to avoid editing the
6376 # instance and then saving the defaults in the instance itself.
6377 hvparams = cluster.FillHV(instance)
6378 beparams = cluster.FillBE(instance)
6379 console_cmd = hyper.GetShellCommandForConsole(instance, hvparams, beparams)
6382 return self.ssh.BuildCmd(node, "root", console_cmd, batch=True, tty=True)
6385 class LUReplaceDisks(LogicalUnit):
6386 """Replace the disks of an instance.
6389 HPATH = "mirrors-replace"
6390 HTYPE = constants.HTYPE_INSTANCE
6391 _OP_REQP = ["instance_name", "mode", "disks"]
6394 def CheckArguments(self):
6395 if not hasattr(self.op, "remote_node"):
6396 self.op.remote_node = None
6397 if not hasattr(self.op, "iallocator"):
6398 self.op.iallocator = None
6399 if not hasattr(self.op, "early_release"):
6400 self.op.early_release = False
6402 TLReplaceDisks.CheckArguments(self.op.mode, self.op.remote_node,
6405 def ExpandNames(self):
6406 self._ExpandAndLockInstance()
6408 if self.op.iallocator is not None:
6409 self.needed_locks[locking.LEVEL_NODE] = locking.ALL_SET
6411 elif self.op.remote_node is not None:
6412 remote_node = _ExpandNodeName(self.cfg, self.op.remote_node)
6413 self.op.remote_node = remote_node
6415 # Warning: do not remove the locking of the new secondary here
6416 # unless DRBD8.AddChildren is changed to work in parallel;
6417 # currently it doesn't since parallel invocations of
6418 # FindUnusedMinor will conflict
6419 self.needed_locks[locking.LEVEL_NODE] = [remote_node]
6420 self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_APPEND
6423 self.needed_locks[locking.LEVEL_NODE] = []
6424 self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
6426 self.replacer = TLReplaceDisks(self, self.op.instance_name, self.op.mode,
6427 self.op.iallocator, self.op.remote_node,
6428 self.op.disks, False, self.op.early_release)
6430 self.tasklets = [self.replacer]
6432 def DeclareLocks(self, level):
6433 # If we're not already locking all nodes in the set we have to declare the
6434 # instance's primary/secondary nodes.
6435 if (level == locking.LEVEL_NODE and
6436 self.needed_locks[locking.LEVEL_NODE] is not locking.ALL_SET):
6437 self._LockInstancesNodes()
6439 def BuildHooksEnv(self):
6442 This runs on the master, the primary and all the secondaries.
6445 instance = self.replacer.instance
6447 "MODE": self.op.mode,
6448 "NEW_SECONDARY": self.op.remote_node,
6449 "OLD_SECONDARY": instance.secondary_nodes[0],
6451 env.update(_BuildInstanceHookEnvByObject(self, instance))
6453 self.cfg.GetMasterNode(),
6454 instance.primary_node,
6456 if self.op.remote_node is not None:
6457 nl.append(self.op.remote_node)
6461 class LUEvacuateNode(LogicalUnit):
6462 """Relocate the secondary instances from a node.
6465 HPATH = "node-evacuate"
6466 HTYPE = constants.HTYPE_NODE
6467 _OP_REQP = ["node_name"]
6470 def CheckArguments(self):
6471 if not hasattr(self.op, "remote_node"):
6472 self.op.remote_node = None
6473 if not hasattr(self.op, "iallocator"):
6474 self.op.iallocator = None
6475 if not hasattr(self.op, "early_release"):
6476 self.op.early_release = False
6478 TLReplaceDisks.CheckArguments(constants.REPLACE_DISK_CHG,
6479 self.op.remote_node,
6482 def ExpandNames(self):
6483 self.op.node_name = _ExpandNodeName(self.cfg, self.op.node_name)
6485 self.needed_locks = {}
6487 # Declare node locks
6488 if self.op.iallocator is not None:
6489 self.needed_locks[locking.LEVEL_NODE] = locking.ALL_SET
6491 elif self.op.remote_node is not None:
6492 self.op.remote_node = _ExpandNodeName(self.cfg, self.op.remote_node)
6494 # Warning: do not remove the locking of the new secondary here
6495 # unless DRBD8.AddChildren is changed to work in parallel;
6496 # currently it doesn't since parallel invocations of
6497 # FindUnusedMinor will conflict
6498 self.needed_locks[locking.LEVEL_NODE] = [self.op.remote_node]
6499 self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_APPEND
6502 raise errors.OpPrereqError("Invalid parameters", errors.ECODE_INVAL)
6504 # Create tasklets for replacing disks for all secondary instances on this
6509 for inst in _GetNodeSecondaryInstances(self.cfg, self.op.node_name):
6510 logging.debug("Replacing disks for instance %s", inst.name)
6511 names.append(inst.name)
6513 replacer = TLReplaceDisks(self, inst.name, constants.REPLACE_DISK_CHG,
6514 self.op.iallocator, self.op.remote_node, [],
6515 True, self.op.early_release)
6516 tasklets.append(replacer)
6518 self.tasklets = tasklets
6519 self.instance_names = names
6521 # Declare instance locks
6522 self.needed_locks[locking.LEVEL_INSTANCE] = self.instance_names
6524 def DeclareLocks(self, level):
6525 # If we're not already locking all nodes in the set we have to declare the
6526 # instance's primary/secondary nodes.
6527 if (level == locking.LEVEL_NODE and
6528 self.needed_locks[locking.LEVEL_NODE] is not locking.ALL_SET):
6529 self._LockInstancesNodes()
6531 def BuildHooksEnv(self):
6534 This runs on the master, the primary and all the secondaries.
6538 "NODE_NAME": self.op.node_name,
6541 nl = [self.cfg.GetMasterNode()]
6543 if self.op.remote_node is not None:
6544 env["NEW_SECONDARY"] = self.op.remote_node
6545 nl.append(self.op.remote_node)
6547 return (env, nl, nl)
6550 class TLReplaceDisks(Tasklet):
6551 """Replaces disks for an instance.
6553 Note: Locking is not within the scope of this class.
6556 def __init__(self, lu, instance_name, mode, iallocator_name, remote_node,
6557 disks, delay_iallocator, early_release):
6558 """Initializes this class.
6561 Tasklet.__init__(self, lu)
6564 self.instance_name = instance_name
6566 self.iallocator_name = iallocator_name
6567 self.remote_node = remote_node
6569 self.delay_iallocator = delay_iallocator
6570 self.early_release = early_release
6573 self.instance = None
6574 self.new_node = None
6575 self.target_node = None
6576 self.other_node = None
6577 self.remote_node_info = None
6578 self.node_secondary_ip = None
6581 def CheckArguments(mode, remote_node, iallocator):
6582 """Helper function for users of this class.
6585 # check for valid parameter combination
6586 if mode == constants.REPLACE_DISK_CHG:
6587 if remote_node is None and iallocator is None:
6588 raise errors.OpPrereqError("When changing the secondary either an"
6589 " iallocator script must be used or the"
6590 " new node given", errors.ECODE_INVAL)
6592 if remote_node is not None and iallocator is not None:
6593 raise errors.OpPrereqError("Give either the iallocator or the new"
6594 " secondary, not both", errors.ECODE_INVAL)
6596 elif remote_node is not None or iallocator is not None:
6597 # Not replacing the secondary
6598 raise errors.OpPrereqError("The iallocator and new node options can"
6599 " only be used when changing the"
6600 " secondary node", errors.ECODE_INVAL)
6603 def _RunAllocator(lu, iallocator_name, instance_name, relocate_from):
6604 """Compute a new secondary node using an IAllocator.
6607 ial = IAllocator(lu.cfg, lu.rpc,
6608 mode=constants.IALLOCATOR_MODE_RELOC,
6610 relocate_from=relocate_from)
6612 ial.Run(iallocator_name)
6615 raise errors.OpPrereqError("Can't compute nodes using iallocator '%s':"
6616 " %s" % (iallocator_name, ial.info),
6619 if len(ial.result) != ial.required_nodes:
6620 raise errors.OpPrereqError("iallocator '%s' returned invalid number"
6621 " of nodes (%s), required %s" %
6623 len(ial.result), ial.required_nodes),
6626 remote_node_name = ial.result[0]
6628 lu.LogInfo("Selected new secondary for instance '%s': %s",
6629 instance_name, remote_node_name)
6631 return remote_node_name
6633 def _FindFaultyDisks(self, node_name):
6634 return _FindFaultyInstanceDisks(self.cfg, self.rpc, self.instance,
6637 def CheckPrereq(self):
6638 """Check prerequisites.
6640 This checks that the instance is in the cluster.
6643 self.instance = instance = self.cfg.GetInstanceInfo(self.instance_name)
6644 assert instance is not None, \
6645 "Cannot retrieve locked instance %s" % self.instance_name
6647 if instance.disk_template != constants.DT_DRBD8:
6648 raise errors.OpPrereqError("Can only run replace disks for DRBD8-based"
6649 " instances", errors.ECODE_INVAL)
6651 if len(instance.secondary_nodes) != 1:
6652 raise errors.OpPrereqError("The instance has a strange layout,"
6653 " expected one secondary but found %d" %
6654 len(instance.secondary_nodes),
6657 if not self.delay_iallocator:
6658 self._CheckPrereq2()
6660 def _CheckPrereq2(self):
6661 """Check prerequisites, second part.
6663 This function should always be part of CheckPrereq. It was separated and is
6664 now called from Exec because during node evacuation iallocator was only
6665 called with an unmodified cluster model, not taking planned changes into
6669 instance = self.instance
6670 secondary_node = instance.secondary_nodes[0]
6672 if self.iallocator_name is None:
6673 remote_node = self.remote_node
6675 remote_node = self._RunAllocator(self.lu, self.iallocator_name,
6676 instance.name, instance.secondary_nodes)
6678 if remote_node is not None:
6679 self.remote_node_info = self.cfg.GetNodeInfo(remote_node)
6680 assert self.remote_node_info is not None, \
6681 "Cannot retrieve locked node %s" % remote_node
6683 self.remote_node_info = None
6685 if remote_node == self.instance.primary_node:
6686 raise errors.OpPrereqError("The specified node is the primary node of"
6687 " the instance.", errors.ECODE_INVAL)
6689 if remote_node == secondary_node:
6690 raise errors.OpPrereqError("The specified node is already the"
6691 " secondary node of the instance.",
6694 if self.disks and self.mode in (constants.REPLACE_DISK_AUTO,
6695 constants.REPLACE_DISK_CHG):
6696 raise errors.OpPrereqError("Cannot specify disks to be replaced",
6699 if self.mode == constants.REPLACE_DISK_AUTO:
6700 faulty_primary = self._FindFaultyDisks(instance.primary_node)
6701 faulty_secondary = self._FindFaultyDisks(secondary_node)
6703 if faulty_primary and faulty_secondary:
6704 raise errors.OpPrereqError("Instance %s has faulty disks on more than"
6705 " one node and can not be repaired"
6706 " automatically" % self.instance_name,
6710 self.disks = faulty_primary
6711 self.target_node = instance.primary_node
6712 self.other_node = secondary_node
6713 check_nodes = [self.target_node, self.other_node]
6714 elif faulty_secondary:
6715 self.disks = faulty_secondary
6716 self.target_node = secondary_node
6717 self.other_node = instance.primary_node
6718 check_nodes = [self.target_node, self.other_node]
6724 # Non-automatic modes
6725 if self.mode == constants.REPLACE_DISK_PRI:
6726 self.target_node = instance.primary_node
6727 self.other_node = secondary_node
6728 check_nodes = [self.target_node, self.other_node]
6730 elif self.mode == constants.REPLACE_DISK_SEC:
6731 self.target_node = secondary_node
6732 self.other_node = instance.primary_node
6733 check_nodes = [self.target_node, self.other_node]
6735 elif self.mode == constants.REPLACE_DISK_CHG:
6736 self.new_node = remote_node
6737 self.other_node = instance.primary_node
6738 self.target_node = secondary_node
6739 check_nodes = [self.new_node, self.other_node]
6741 _CheckNodeNotDrained(self.lu, remote_node)
6743 old_node_info = self.cfg.GetNodeInfo(secondary_node)
6744 assert old_node_info is not None
6745 if old_node_info.offline and not self.early_release:
6746 # doesn't make sense to delay the release
6747 self.early_release = True
6748 self.lu.LogInfo("Old secondary %s is offline, automatically enabling"
6749 " early-release mode", secondary_node)
6752 raise errors.ProgrammerError("Unhandled disk replace mode (%s)" %
6755 # If not specified all disks should be replaced
6757 self.disks = range(len(self.instance.disks))
6759 for node in check_nodes:
6760 _CheckNodeOnline(self.lu, node)
6762 # Check whether disks are valid
6763 for disk_idx in self.disks:
6764 instance.FindDisk(disk_idx)
6766 # Get secondary node IP addresses
6769 for node_name in [self.target_node, self.other_node, self.new_node]:
6770 if node_name is not None:
6771 node_2nd_ip[node_name] = self.cfg.GetNodeInfo(node_name).secondary_ip
6773 self.node_secondary_ip = node_2nd_ip
6775 def Exec(self, feedback_fn):
6776 """Execute disk replacement.
6778 This dispatches the disk replacement to the appropriate handler.
6781 if self.delay_iallocator:
6782 self._CheckPrereq2()
6785 feedback_fn("No disks need replacement")
6788 feedback_fn("Replacing disk(s) %s for %s" %
6789 (utils.CommaJoin(self.disks), self.instance.name))
6791 activate_disks = (not self.instance.admin_up)
6793 # Activate the instance disks if we're replacing them on a down instance
6795 _StartInstanceDisks(self.lu, self.instance, True)
6798 # Should we replace the secondary node?
6799 if self.new_node is not None:
6800 fn = self._ExecDrbd8Secondary
6802 fn = self._ExecDrbd8DiskOnly
6804 return fn(feedback_fn)
6807 # Deactivate the instance disks if we're replacing them on a
6810 _SafeShutdownInstanceDisks(self.lu, self.instance)
6812 def _CheckVolumeGroup(self, nodes):
6813 self.lu.LogInfo("Checking volume groups")
6815 vgname = self.cfg.GetVGName()
6817 # Make sure volume group exists on all involved nodes
6818 results = self.rpc.call_vg_list(nodes)
6820 raise errors.OpExecError("Can't list volume groups on the nodes")
6824 res.Raise("Error checking node %s" % node)
6825 if vgname not in res.payload:
6826 raise errors.OpExecError("Volume group '%s' not found on node %s" %
6829 def _CheckDisksExistence(self, nodes):
6830 # Check disk existence
6831 for idx, dev in enumerate(self.instance.disks):
6832 if idx not in self.disks:
6836 self.lu.LogInfo("Checking disk/%d on %s" % (idx, node))
6837 self.cfg.SetDiskID(dev, node)
6839 result = self.rpc.call_blockdev_find(node, dev)
6841 msg = result.fail_msg
6842 if msg or not result.payload:
6844 msg = "disk not found"
6845 raise errors.OpExecError("Can't find disk/%d on node %s: %s" %
6848 def _CheckDisksConsistency(self, node_name, on_primary, ldisk):
6849 for idx, dev in enumerate(self.instance.disks):
6850 if idx not in self.disks:
6853 self.lu.LogInfo("Checking disk/%d consistency on node %s" %
6856 if not _CheckDiskConsistency(self.lu, dev, node_name, on_primary,
6858 raise errors.OpExecError("Node %s has degraded storage, unsafe to"
6859 " replace disks for instance %s" %
6860 (node_name, self.instance.name))
6862 def _CreateNewStorage(self, node_name):
6863 vgname = self.cfg.GetVGName()
6866 for idx, dev in enumerate(self.instance.disks):
6867 if idx not in self.disks:
6870 self.lu.LogInfo("Adding storage on %s for disk/%d" % (node_name, idx))
6872 self.cfg.SetDiskID(dev, node_name)
6874 lv_names = [".disk%d_%s" % (idx, suffix) for suffix in ["data", "meta"]]
6875 names = _GenerateUniqueNames(self.lu, lv_names)
6877 lv_data = objects.Disk(dev_type=constants.LD_LV, size=dev.size,
6878 logical_id=(vgname, names[0]))
6879 lv_meta = objects.Disk(dev_type=constants.LD_LV, size=128,
6880 logical_id=(vgname, names[1]))
6882 new_lvs = [lv_data, lv_meta]
6883 old_lvs = dev.children
6884 iv_names[dev.iv_name] = (dev, old_lvs, new_lvs)
6886 # we pass force_create=True to force the LVM creation
6887 for new_lv in new_lvs:
6888 _CreateBlockDev(self.lu, node_name, self.instance, new_lv, True,
6889 _GetInstanceInfoText(self.instance), False)
6893 def _CheckDevices(self, node_name, iv_names):
6894 for name, (dev, _, _) in iv_names.iteritems():
6895 self.cfg.SetDiskID(dev, node_name)
6897 result = self.rpc.call_blockdev_find(node_name, dev)
6899 msg = result.fail_msg
6900 if msg or not result.payload:
6902 msg = "disk not found"
6903 raise errors.OpExecError("Can't find DRBD device %s: %s" %
6906 if result.payload.is_degraded:
6907 raise errors.OpExecError("DRBD device %s is degraded!" % name)
6909 def _RemoveOldStorage(self, node_name, iv_names):
6910 for name, (_, old_lvs, _) in iv_names.iteritems():
6911 self.lu.LogInfo("Remove logical volumes for %s" % name)
6914 self.cfg.SetDiskID(lv, node_name)
6916 msg = self.rpc.call_blockdev_remove(node_name, lv).fail_msg
6918 self.lu.LogWarning("Can't remove old LV: %s" % msg,
6919 hint="remove unused LVs manually")
6921 def _ReleaseNodeLock(self, node_name):
6922 """Releases the lock for a given node."""
6923 self.lu.context.glm.release(locking.LEVEL_NODE, node_name)
6925 def _ExecDrbd8DiskOnly(self, feedback_fn):
6926 """Replace a disk on the primary or secondary for DRBD 8.
6928 The algorithm for replace is quite complicated:
6930 1. for each disk to be replaced:
6932 1. create new LVs on the target node with unique names
6933 1. detach old LVs from the drbd device
6934 1. rename old LVs to name_replaced.<time_t>
6935 1. rename new LVs to old LVs
6936 1. attach the new LVs (with the old names now) to the drbd device
6938 1. wait for sync across all devices
6940 1. for each modified disk:
6942 1. remove old LVs (which have the name name_replaces.<time_t>)
6944 Failures are not very well handled.
6949 # Step: check device activation
6950 self.lu.LogStep(1, steps_total, "Check device existence")
6951 self._CheckDisksExistence([self.other_node, self.target_node])
6952 self._CheckVolumeGroup([self.target_node, self.other_node])
6954 # Step: check other node consistency
6955 self.lu.LogStep(2, steps_total, "Check peer consistency")
6956 self._CheckDisksConsistency(self.other_node,
6957 self.other_node == self.instance.primary_node,
6960 # Step: create new storage
6961 self.lu.LogStep(3, steps_total, "Allocate new storage")
6962 iv_names = self._CreateNewStorage(self.target_node)
6964 # Step: for each lv, detach+rename*2+attach
6965 self.lu.LogStep(4, steps_total, "Changing drbd configuration")
6966 for dev, old_lvs, new_lvs in iv_names.itervalues():
6967 self.lu.LogInfo("Detaching %s drbd from local storage" % dev.iv_name)
6969 result = self.rpc.call_blockdev_removechildren(self.target_node, dev,
6971 result.Raise("Can't detach drbd from local storage on node"
6972 " %s for device %s" % (self.target_node, dev.iv_name))
6974 #cfg.Update(instance)
6976 # ok, we created the new LVs, so now we know we have the needed
6977 # storage; as such, we proceed on the target node to rename
6978 # old_lv to _old, and new_lv to old_lv; note that we rename LVs
6979 # using the assumption that logical_id == physical_id (which in
6980 # turn is the unique_id on that node)
6982 # FIXME(iustin): use a better name for the replaced LVs
6983 temp_suffix = int(time.time())
6984 ren_fn = lambda d, suff: (d.physical_id[0],
6985 d.physical_id[1] + "_replaced-%s" % suff)
6987 # Build the rename list based on what LVs exist on the node
6988 rename_old_to_new = []
6989 for to_ren in old_lvs:
6990 result = self.rpc.call_blockdev_find(self.target_node, to_ren)
6991 if not result.fail_msg and result.payload:
6993 rename_old_to_new.append((to_ren, ren_fn(to_ren, temp_suffix)))
6995 self.lu.LogInfo("Renaming the old LVs on the target node")
6996 result = self.rpc.call_blockdev_rename(self.target_node,
6998 result.Raise("Can't rename old LVs on node %s" % self.target_node)
7000 # Now we rename the new LVs to the old LVs
7001 self.lu.LogInfo("Renaming the new LVs on the target node")
7002 rename_new_to_old = [(new, old.physical_id)
7003 for old, new in zip(old_lvs, new_lvs)]
7004 result = self.rpc.call_blockdev_rename(self.target_node,
7006 result.Raise("Can't rename new LVs on node %s" % self.target_node)
7008 for old, new in zip(old_lvs, new_lvs):
7009 new.logical_id = old.logical_id
7010 self.cfg.SetDiskID(new, self.target_node)
7012 for disk in old_lvs:
7013 disk.logical_id = ren_fn(disk, temp_suffix)
7014 self.cfg.SetDiskID(disk, self.target_node)
7016 # Now that the new lvs have the old name, we can add them to the device
7017 self.lu.LogInfo("Adding new mirror component on %s" % self.target_node)
7018 result = self.rpc.call_blockdev_addchildren(self.target_node, dev,
7020 msg = result.fail_msg
7022 for new_lv in new_lvs:
7023 msg2 = self.rpc.call_blockdev_remove(self.target_node,
7026 self.lu.LogWarning("Can't rollback device %s: %s", dev, msg2,
7027 hint=("cleanup manually the unused logical"
7029 raise errors.OpExecError("Can't add local storage to drbd: %s" % msg)
7031 dev.children = new_lvs
7033 self.cfg.Update(self.instance, feedback_fn)
7036 if self.early_release:
7037 self.lu.LogStep(cstep, steps_total, "Removing old storage")
7039 self._RemoveOldStorage(self.target_node, iv_names)
7040 # WARNING: we release both node locks here, do not do other RPCs
7041 # than WaitForSync to the primary node
7042 self._ReleaseNodeLock([self.target_node, self.other_node])
7045 # This can fail as the old devices are degraded and _WaitForSync
7046 # does a combined result over all disks, so we don't check its return value
7047 self.lu.LogStep(cstep, steps_total, "Sync devices")
7049 _WaitForSync(self.lu, self.instance)
7051 # Check all devices manually
7052 self._CheckDevices(self.instance.primary_node, iv_names)
7054 # Step: remove old storage
7055 if not self.early_release:
7056 self.lu.LogStep(cstep, steps_total, "Removing old storage")
7058 self._RemoveOldStorage(self.target_node, iv_names)
7060 def _ExecDrbd8Secondary(self, feedback_fn):
7061 """Replace the secondary node for DRBD 8.
7063 The algorithm for replace is quite complicated:
7064 - for all disks of the instance:
7065 - create new LVs on the new node with same names
7066 - shutdown the drbd device on the old secondary
7067 - disconnect the drbd network on the primary
7068 - create the drbd device on the new secondary
7069 - network attach the drbd on the primary, using an artifice:
7070 the drbd code for Attach() will connect to the network if it
7071 finds a device which is connected to the good local disks but
7073 - wait for sync across all devices
7074 - remove all disks from the old secondary
7076 Failures are not very well handled.
7081 # Step: check device activation
7082 self.lu.LogStep(1, steps_total, "Check device existence")
7083 self._CheckDisksExistence([self.instance.primary_node])
7084 self._CheckVolumeGroup([self.instance.primary_node])
7086 # Step: check other node consistency
7087 self.lu.LogStep(2, steps_total, "Check peer consistency")
7088 self._CheckDisksConsistency(self.instance.primary_node, True, True)
7090 # Step: create new storage
7091 self.lu.LogStep(3, steps_total, "Allocate new storage")
7092 for idx, dev in enumerate(self.instance.disks):
7093 self.lu.LogInfo("Adding new local storage on %s for disk/%d" %
7094 (self.new_node, idx))
7095 # we pass force_create=True to force LVM creation
7096 for new_lv in dev.children:
7097 _CreateBlockDev(self.lu, self.new_node, self.instance, new_lv, True,
7098 _GetInstanceInfoText(self.instance), False)
7100 # Step 4: dbrd minors and drbd setups changes
7101 # after this, we must manually remove the drbd minors on both the
7102 # error and the success paths
7103 self.lu.LogStep(4, steps_total, "Changing drbd configuration")
7104 minors = self.cfg.AllocateDRBDMinor([self.new_node
7105 for dev in self.instance.disks],
7107 logging.debug("Allocated minors %r", minors)
7110 for idx, (dev, new_minor) in enumerate(zip(self.instance.disks, minors)):
7111 self.lu.LogInfo("activating a new drbd on %s for disk/%d" %
7112 (self.new_node, idx))
7113 # create new devices on new_node; note that we create two IDs:
7114 # one without port, so the drbd will be activated without
7115 # networking information on the new node at this stage, and one
7116 # with network, for the latter activation in step 4
7117 (o_node1, o_node2, o_port, o_minor1, o_minor2, o_secret) = dev.logical_id
7118 if self.instance.primary_node == o_node1:
7121 assert self.instance.primary_node == o_node2, "Three-node instance?"
7124 new_alone_id = (self.instance.primary_node, self.new_node, None,
7125 p_minor, new_minor, o_secret)
7126 new_net_id = (self.instance.primary_node, self.new_node, o_port,
7127 p_minor, new_minor, o_secret)
7129 iv_names[idx] = (dev, dev.children, new_net_id)
7130 logging.debug("Allocated new_minor: %s, new_logical_id: %s", new_minor,
7132 new_drbd = objects.Disk(dev_type=constants.LD_DRBD8,
7133 logical_id=new_alone_id,
7134 children=dev.children,
7137 _CreateSingleBlockDev(self.lu, self.new_node, self.instance, new_drbd,
7138 _GetInstanceInfoText(self.instance), False)
7139 except errors.GenericError:
7140 self.cfg.ReleaseDRBDMinors(self.instance.name)
7143 # We have new devices, shutdown the drbd on the old secondary
7144 for idx, dev in enumerate(self.instance.disks):
7145 self.lu.LogInfo("Shutting down drbd for disk/%d on old node" % idx)
7146 self.cfg.SetDiskID(dev, self.target_node)
7147 msg = self.rpc.call_blockdev_shutdown(self.target_node, dev).fail_msg
7149 self.lu.LogWarning("Failed to shutdown drbd for disk/%d on old"
7150 "node: %s" % (idx, msg),
7151 hint=("Please cleanup this device manually as"
7152 " soon as possible"))
7154 self.lu.LogInfo("Detaching primary drbds from the network (=> standalone)")
7155 result = self.rpc.call_drbd_disconnect_net([self.instance.primary_node],
7156 self.node_secondary_ip,
7157 self.instance.disks)\
7158 [self.instance.primary_node]
7160 msg = result.fail_msg
7162 # detaches didn't succeed (unlikely)
7163 self.cfg.ReleaseDRBDMinors(self.instance.name)
7164 raise errors.OpExecError("Can't detach the disks from the network on"
7165 " old node: %s" % (msg,))
7167 # if we managed to detach at least one, we update all the disks of
7168 # the instance to point to the new secondary
7169 self.lu.LogInfo("Updating instance configuration")
7170 for dev, _, new_logical_id in iv_names.itervalues():
7171 dev.logical_id = new_logical_id
7172 self.cfg.SetDiskID(dev, self.instance.primary_node)
7174 self.cfg.Update(self.instance, feedback_fn)
7176 # and now perform the drbd attach
7177 self.lu.LogInfo("Attaching primary drbds to new secondary"
7178 " (standalone => connected)")
7179 result = self.rpc.call_drbd_attach_net([self.instance.primary_node,
7181 self.node_secondary_ip,
7182 self.instance.disks,
7185 for to_node, to_result in result.items():
7186 msg = to_result.fail_msg
7188 self.lu.LogWarning("Can't attach drbd disks on node %s: %s",
7190 hint=("please do a gnt-instance info to see the"
7191 " status of disks"))
7193 if self.early_release:
7194 self.lu.LogStep(cstep, steps_total, "Removing old storage")
7196 self._RemoveOldStorage(self.target_node, iv_names)
7197 # WARNING: we release all node locks here, do not do other RPCs
7198 # than WaitForSync to the primary node
7199 self._ReleaseNodeLock([self.instance.primary_node,
7204 # This can fail as the old devices are degraded and _WaitForSync
7205 # does a combined result over all disks, so we don't check its return value
7206 self.lu.LogStep(cstep, steps_total, "Sync devices")
7208 _WaitForSync(self.lu, self.instance)
7210 # Check all devices manually
7211 self._CheckDevices(self.instance.primary_node, iv_names)
7213 # Step: remove old storage
7214 if not self.early_release:
7215 self.lu.LogStep(cstep, steps_total, "Removing old storage")
7216 self._RemoveOldStorage(self.target_node, iv_names)
7219 class LURepairNodeStorage(NoHooksLU):
7220 """Repairs the volume group on a node.
7223 _OP_REQP = ["node_name"]
7226 def CheckArguments(self):
7227 self.op.node_name = _ExpandNodeName(self.cfg, self.op.node_name)
7229 def ExpandNames(self):
7230 self.needed_locks = {
7231 locking.LEVEL_NODE: [self.op.node_name],
7234 def _CheckFaultyDisks(self, instance, node_name):
7235 """Ensure faulty disks abort the opcode or at least warn."""
7237 if _FindFaultyInstanceDisks(self.cfg, self.rpc, instance,
7239 raise errors.OpPrereqError("Instance '%s' has faulty disks on"
7240 " node '%s'" % (instance.name, node_name),
7242 except errors.OpPrereqError, err:
7243 if self.op.ignore_consistency:
7244 self.proc.LogWarning(str(err.args[0]))
7248 def CheckPrereq(self):
7249 """Check prerequisites.
7252 storage_type = self.op.storage_type
7254 if (constants.SO_FIX_CONSISTENCY not in
7255 constants.VALID_STORAGE_OPERATIONS.get(storage_type, [])):
7256 raise errors.OpPrereqError("Storage units of type '%s' can not be"
7257 " repaired" % storage_type,
7260 # Check whether any instance on this node has faulty disks
7261 for inst in _GetNodeInstances(self.cfg, self.op.node_name):
7262 if not inst.admin_up:
7264 check_nodes = set(inst.all_nodes)
7265 check_nodes.discard(self.op.node_name)
7266 for inst_node_name in check_nodes:
7267 self._CheckFaultyDisks(inst, inst_node_name)
7269 def Exec(self, feedback_fn):
7270 feedback_fn("Repairing storage unit '%s' on %s ..." %
7271 (self.op.name, self.op.node_name))
7273 st_args = _GetStorageTypeArgs(self.cfg, self.op.storage_type)
7274 result = self.rpc.call_storage_execute(self.op.node_name,
7275 self.op.storage_type, st_args,
7277 constants.SO_FIX_CONSISTENCY)
7278 result.Raise("Failed to repair storage unit '%s' on %s" %
7279 (self.op.name, self.op.node_name))
7282 class LUNodeEvacuationStrategy(NoHooksLU):
7283 """Computes the node evacuation strategy.
7286 _OP_REQP = ["nodes"]
7289 def CheckArguments(self):
7290 if not hasattr(self.op, "remote_node"):
7291 self.op.remote_node = None
7292 if not hasattr(self.op, "iallocator"):
7293 self.op.iallocator = None
7294 if self.op.remote_node is not None and self.op.iallocator is not None:
7295 raise errors.OpPrereqError("Give either the iallocator or the new"
7296 " secondary, not both", errors.ECODE_INVAL)
7298 def ExpandNames(self):
7299 self.op.nodes = _GetWantedNodes(self, self.op.nodes)
7300 self.needed_locks = locks = {}
7301 if self.op.remote_node is None:
7302 locks[locking.LEVEL_NODE] = locking.ALL_SET
7304 self.op.remote_node = _ExpandNodeName(self.cfg, self.op.remote_node)
7305 locks[locking.LEVEL_NODE] = self.op.nodes + [self.op.remote_node]
7307 def CheckPrereq(self):
7310 def Exec(self, feedback_fn):
7311 if self.op.remote_node is not None:
7313 for node in self.op.nodes:
7314 instances.extend(_GetNodeSecondaryInstances(self.cfg, node))
7317 if i.primary_node == self.op.remote_node:
7318 raise errors.OpPrereqError("Node %s is the primary node of"
7319 " instance %s, cannot use it as"
7321 (self.op.remote_node, i.name),
7323 result.append([i.name, self.op.remote_node])
7325 ial = IAllocator(self.cfg, self.rpc,
7326 mode=constants.IALLOCATOR_MODE_MEVAC,
7327 evac_nodes=self.op.nodes)
7328 ial.Run(self.op.iallocator, validate=True)
7330 raise errors.OpExecError("No valid evacuation solution: %s" % ial.info,
7336 class LUGrowDisk(LogicalUnit):
7337 """Grow a disk of an instance.
7341 HTYPE = constants.HTYPE_INSTANCE
7342 _OP_REQP = ["instance_name", "disk", "amount", "wait_for_sync"]
7345 def ExpandNames(self):
7346 self._ExpandAndLockInstance()
7347 self.needed_locks[locking.LEVEL_NODE] = []
7348 self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
7350 def DeclareLocks(self, level):
7351 if level == locking.LEVEL_NODE:
7352 self._LockInstancesNodes()
7354 def BuildHooksEnv(self):
7357 This runs on the master, the primary and all the secondaries.
7361 "DISK": self.op.disk,
7362 "AMOUNT": self.op.amount,
7364 env.update(_BuildInstanceHookEnvByObject(self, self.instance))
7365 nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
7368 def CheckPrereq(self):
7369 """Check prerequisites.
7371 This checks that the instance is in the cluster.
7374 instance = self.cfg.GetInstanceInfo(self.op.instance_name)
7375 assert instance is not None, \
7376 "Cannot retrieve locked instance %s" % self.op.instance_name
7377 nodenames = list(instance.all_nodes)
7378 for node in nodenames:
7379 _CheckNodeOnline(self, node)
7382 self.instance = instance
7384 if instance.disk_template not in (constants.DT_PLAIN, constants.DT_DRBD8):
7385 raise errors.OpPrereqError("Instance's disk layout does not support"
7386 " growing.", errors.ECODE_INVAL)
7388 self.disk = instance.FindDisk(self.op.disk)
7390 nodeinfo = self.rpc.call_node_info(nodenames, self.cfg.GetVGName(),
7391 instance.hypervisor)
7392 for node in nodenames:
7393 info = nodeinfo[node]
7394 info.Raise("Cannot get current information from node %s" % node)
7395 vg_free = info.payload.get('vg_free', None)
7396 if not isinstance(vg_free, int):
7397 raise errors.OpPrereqError("Can't compute free disk space on"
7398 " node %s" % node, errors.ECODE_ENVIRON)
7399 if self.op.amount > vg_free:
7400 raise errors.OpPrereqError("Not enough disk space on target node %s:"
7401 " %d MiB available, %d MiB required" %
7402 (node, vg_free, self.op.amount),
7405 def Exec(self, feedback_fn):
7406 """Execute disk grow.
7409 instance = self.instance
7411 for node in instance.all_nodes:
7412 self.cfg.SetDiskID(disk, node)
7413 result = self.rpc.call_blockdev_grow(node, disk, self.op.amount)
7414 result.Raise("Grow request failed to node %s" % node)
7416 # TODO: Rewrite code to work properly
7417 # DRBD goes into sync mode for a short amount of time after executing the
7418 # "resize" command. DRBD 8.x below version 8.0.13 contains a bug whereby
7419 # calling "resize" in sync mode fails. Sleeping for a short amount of
7420 # time is a work-around.
7423 disk.RecordGrow(self.op.amount)
7424 self.cfg.Update(instance, feedback_fn)
7425 if self.op.wait_for_sync:
7426 disk_abort = not _WaitForSync(self, instance)
7428 self.proc.LogWarning("Warning: disk sync-ing has not returned a good"
7429 " status.\nPlease check the instance.")
7432 class LUQueryInstanceData(NoHooksLU):
7433 """Query runtime instance data.
7436 _OP_REQP = ["instances", "static"]
7439 def ExpandNames(self):
7440 self.needed_locks = {}
7441 self.share_locks = dict.fromkeys(locking.LEVELS, 1)
7443 if not isinstance(self.op.instances, list):
7444 raise errors.OpPrereqError("Invalid argument type 'instances'",
7447 if self.op.instances:
7448 self.wanted_names = []
7449 for name in self.op.instances:
7450 full_name = _ExpandInstanceName(self.cfg, name)
7451 self.wanted_names.append(full_name)
7452 self.needed_locks[locking.LEVEL_INSTANCE] = self.wanted_names
7454 self.wanted_names = None
7455 self.needed_locks[locking.LEVEL_INSTANCE] = locking.ALL_SET
7457 self.needed_locks[locking.LEVEL_NODE] = []
7458 self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
7460 def DeclareLocks(self, level):
7461 if level == locking.LEVEL_NODE:
7462 self._LockInstancesNodes()
7464 def CheckPrereq(self):
7465 """Check prerequisites.
7467 This only checks the optional instance list against the existing names.
7470 if self.wanted_names is None:
7471 self.wanted_names = self.acquired_locks[locking.LEVEL_INSTANCE]
7473 self.wanted_instances = [self.cfg.GetInstanceInfo(name) for name
7474 in self.wanted_names]
7477 def _ComputeBlockdevStatus(self, node, instance_name, dev):
7478 """Returns the status of a block device
7481 if self.op.static or not node:
7484 self.cfg.SetDiskID(dev, node)
7486 result = self.rpc.call_blockdev_find(node, dev)
7490 result.Raise("Can't compute disk status for %s" % instance_name)
7492 status = result.payload
7496 return (status.dev_path, status.major, status.minor,
7497 status.sync_percent, status.estimated_time,
7498 status.is_degraded, status.ldisk_status)
7500 def _ComputeDiskStatus(self, instance, snode, dev):
7501 """Compute block device status.
7504 if dev.dev_type in constants.LDS_DRBD:
7505 # we change the snode then (otherwise we use the one passed in)
7506 if dev.logical_id[0] == instance.primary_node:
7507 snode = dev.logical_id[1]
7509 snode = dev.logical_id[0]
7511 dev_pstatus = self._ComputeBlockdevStatus(instance.primary_node,
7513 dev_sstatus = self._ComputeBlockdevStatus(snode, instance.name, dev)
7516 dev_children = [self._ComputeDiskStatus(instance, snode, child)
7517 for child in dev.children]
7522 "iv_name": dev.iv_name,
7523 "dev_type": dev.dev_type,
7524 "logical_id": dev.logical_id,
7525 "physical_id": dev.physical_id,
7526 "pstatus": dev_pstatus,
7527 "sstatus": dev_sstatus,
7528 "children": dev_children,
7535 def Exec(self, feedback_fn):
7536 """Gather and return data"""
7539 cluster = self.cfg.GetClusterInfo()
7541 for instance in self.wanted_instances:
7542 if not self.op.static:
7543 remote_info = self.rpc.call_instance_info(instance.primary_node,
7545 instance.hypervisor)
7546 remote_info.Raise("Error checking node %s" % instance.primary_node)
7547 remote_info = remote_info.payload
7548 if remote_info and "state" in remote_info:
7551 remote_state = "down"
7554 if instance.admin_up:
7557 config_state = "down"
7559 disks = [self._ComputeDiskStatus(instance, None, device)
7560 for device in instance.disks]
7563 "name": instance.name,
7564 "config_state": config_state,
7565 "run_state": remote_state,
7566 "pnode": instance.primary_node,
7567 "snodes": instance.secondary_nodes,
7569 # this happens to be the same format used for hooks
7570 "nics": _NICListToTuple(self, instance.nics),
7572 "hypervisor": instance.hypervisor,
7573 "network_port": instance.network_port,
7574 "hv_instance": instance.hvparams,
7575 "hv_actual": cluster.FillHV(instance, skip_globals=True),
7576 "be_instance": instance.beparams,
7577 "be_actual": cluster.FillBE(instance),
7578 "serial_no": instance.serial_no,
7579 "mtime": instance.mtime,
7580 "ctime": instance.ctime,
7581 "uuid": instance.uuid,
7584 result[instance.name] = idict
7589 class LUSetInstanceParams(LogicalUnit):
7590 """Modifies an instances's parameters.
7593 HPATH = "instance-modify"
7594 HTYPE = constants.HTYPE_INSTANCE
7595 _OP_REQP = ["instance_name"]
7598 def CheckArguments(self):
7599 if not hasattr(self.op, 'nics'):
7601 if not hasattr(self.op, 'disks'):
7603 if not hasattr(self.op, 'beparams'):
7604 self.op.beparams = {}
7605 if not hasattr(self.op, 'hvparams'):
7606 self.op.hvparams = {}
7607 self.op.force = getattr(self.op, "force", False)
7608 if not (self.op.nics or self.op.disks or
7609 self.op.hvparams or self.op.beparams):
7610 raise errors.OpPrereqError("No changes submitted", errors.ECODE_INVAL)
7612 if self.op.hvparams:
7613 _CheckGlobalHvParams(self.op.hvparams)
7617 for disk_op, disk_dict in self.op.disks:
7618 if disk_op == constants.DDM_REMOVE:
7621 elif disk_op == constants.DDM_ADD:
7624 if not isinstance(disk_op, int):
7625 raise errors.OpPrereqError("Invalid disk index", errors.ECODE_INVAL)
7626 if not isinstance(disk_dict, dict):
7627 msg = "Invalid disk value: expected dict, got '%s'" % disk_dict
7628 raise errors.OpPrereqError(msg, errors.ECODE_INVAL)
7630 if disk_op == constants.DDM_ADD:
7631 mode = disk_dict.setdefault('mode', constants.DISK_RDWR)
7632 if mode not in constants.DISK_ACCESS_SET:
7633 raise errors.OpPrereqError("Invalid disk access mode '%s'" % mode,
7635 size = disk_dict.get('size', None)
7637 raise errors.OpPrereqError("Required disk parameter size missing",
7641 except (TypeError, ValueError), err:
7642 raise errors.OpPrereqError("Invalid disk size parameter: %s" %
7643 str(err), errors.ECODE_INVAL)
7644 disk_dict['size'] = size
7646 # modification of disk
7647 if 'size' in disk_dict:
7648 raise errors.OpPrereqError("Disk size change not possible, use"
7649 " grow-disk", errors.ECODE_INVAL)
7651 if disk_addremove > 1:
7652 raise errors.OpPrereqError("Only one disk add or remove operation"
7653 " supported at a time", errors.ECODE_INVAL)
7657 for nic_op, nic_dict in self.op.nics:
7658 if nic_op == constants.DDM_REMOVE:
7661 elif nic_op == constants.DDM_ADD:
7664 if not isinstance(nic_op, int):
7665 raise errors.OpPrereqError("Invalid nic index", errors.ECODE_INVAL)
7666 if not isinstance(nic_dict, dict):
7667 msg = "Invalid nic value: expected dict, got '%s'" % nic_dict
7668 raise errors.OpPrereqError(msg, errors.ECODE_INVAL)
7670 # nic_dict should be a dict
7671 nic_ip = nic_dict.get('ip', None)
7672 if nic_ip is not None:
7673 if nic_ip.lower() == constants.VALUE_NONE:
7674 nic_dict['ip'] = None
7676 if not utils.IsValidIP(nic_ip):
7677 raise errors.OpPrereqError("Invalid IP address '%s'" % nic_ip,
7680 nic_bridge = nic_dict.get('bridge', None)
7681 nic_link = nic_dict.get('link', None)
7682 if nic_bridge and nic_link:
7683 raise errors.OpPrereqError("Cannot pass 'bridge' and 'link'"
7684 " at the same time", errors.ECODE_INVAL)
7685 elif nic_bridge and nic_bridge.lower() == constants.VALUE_NONE:
7686 nic_dict['bridge'] = None
7687 elif nic_link and nic_link.lower() == constants.VALUE_NONE:
7688 nic_dict['link'] = None
7690 if nic_op == constants.DDM_ADD:
7691 nic_mac = nic_dict.get('mac', None)
7693 nic_dict['mac'] = constants.VALUE_AUTO
7695 if 'mac' in nic_dict:
7696 nic_mac = nic_dict['mac']
7697 if nic_mac not in (constants.VALUE_AUTO, constants.VALUE_GENERATE):
7698 nic_mac = utils.NormalizeAndValidateMac(nic_mac)
7700 if nic_op != constants.DDM_ADD and nic_mac == constants.VALUE_AUTO:
7701 raise errors.OpPrereqError("'auto' is not a valid MAC address when"
7702 " modifying an existing nic",
7705 if nic_addremove > 1:
7706 raise errors.OpPrereqError("Only one NIC add or remove operation"
7707 " supported at a time", errors.ECODE_INVAL)
7709 def ExpandNames(self):
7710 self._ExpandAndLockInstance()
7711 self.needed_locks[locking.LEVEL_NODE] = []
7712 self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
7714 def DeclareLocks(self, level):
7715 if level == locking.LEVEL_NODE:
7716 self._LockInstancesNodes()
7718 def BuildHooksEnv(self):
7721 This runs on the master, primary and secondaries.
7725 if constants.BE_MEMORY in self.be_new:
7726 args['memory'] = self.be_new[constants.BE_MEMORY]
7727 if constants.BE_VCPUS in self.be_new:
7728 args['vcpus'] = self.be_new[constants.BE_VCPUS]
7729 # TODO: export disk changes. Note: _BuildInstanceHookEnv* don't export disk
7730 # information at all.
7733 nic_override = dict(self.op.nics)
7734 c_nicparams = self.cluster.nicparams[constants.PP_DEFAULT]
7735 for idx, nic in enumerate(self.instance.nics):
7736 if idx in nic_override:
7737 this_nic_override = nic_override[idx]
7739 this_nic_override = {}
7740 if 'ip' in this_nic_override:
7741 ip = this_nic_override['ip']
7744 if 'mac' in this_nic_override:
7745 mac = this_nic_override['mac']
7748 if idx in self.nic_pnew:
7749 nicparams = self.nic_pnew[idx]
7751 nicparams = objects.FillDict(c_nicparams, nic.nicparams)
7752 mode = nicparams[constants.NIC_MODE]
7753 link = nicparams[constants.NIC_LINK]
7754 args['nics'].append((ip, mac, mode, link))
7755 if constants.DDM_ADD in nic_override:
7756 ip = nic_override[constants.DDM_ADD].get('ip', None)
7757 mac = nic_override[constants.DDM_ADD]['mac']
7758 nicparams = self.nic_pnew[constants.DDM_ADD]
7759 mode = nicparams[constants.NIC_MODE]
7760 link = nicparams[constants.NIC_LINK]
7761 args['nics'].append((ip, mac, mode, link))
7762 elif constants.DDM_REMOVE in nic_override:
7763 del args['nics'][-1]
7765 env = _BuildInstanceHookEnvByObject(self, self.instance, override=args)
7766 nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
7770 def _GetUpdatedParams(old_params, update_dict,
7771 default_values, parameter_types):
7772 """Return the new params dict for the given params.
7774 @type old_params: dict
7775 @param old_params: old parameters
7776 @type update_dict: dict
7777 @param update_dict: dict containing new parameter values,
7778 or constants.VALUE_DEFAULT to reset the
7779 parameter to its default value
7780 @type default_values: dict
7781 @param default_values: default values for the filled parameters
7782 @type parameter_types: dict
7783 @param parameter_types: dict mapping target dict keys to types
7784 in constants.ENFORCEABLE_TYPES
7785 @rtype: (dict, dict)
7786 @return: (new_parameters, filled_parameters)
7789 params_copy = copy.deepcopy(old_params)
7790 for key, val in update_dict.iteritems():
7791 if val == constants.VALUE_DEFAULT:
7793 del params_copy[key]
7797 params_copy[key] = val
7798 utils.ForceDictType(params_copy, parameter_types)
7799 params_filled = objects.FillDict(default_values, params_copy)
7800 return (params_copy, params_filled)
7802 def CheckPrereq(self):
7803 """Check prerequisites.
7805 This only checks the instance list against the existing names.
7808 self.force = self.op.force
7810 # checking the new params on the primary/secondary nodes
7812 instance = self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
7813 cluster = self.cluster = self.cfg.GetClusterInfo()
7814 assert self.instance is not None, \
7815 "Cannot retrieve locked instance %s" % self.op.instance_name
7816 pnode = instance.primary_node
7817 nodelist = list(instance.all_nodes)
7819 # hvparams processing
7820 if self.op.hvparams:
7821 i_hvdict, hv_new = self._GetUpdatedParams(
7822 instance.hvparams, self.op.hvparams,
7823 cluster.hvparams[instance.hypervisor],
7824 constants.HVS_PARAMETER_TYPES)
7826 hypervisor.GetHypervisor(
7827 instance.hypervisor).CheckParameterSyntax(hv_new)
7828 _CheckHVParams(self, nodelist, instance.hypervisor, hv_new)
7829 self.hv_new = hv_new # the new actual values
7830 self.hv_inst = i_hvdict # the new dict (without defaults)
7832 self.hv_new = self.hv_inst = {}
7834 # beparams processing
7835 if self.op.beparams:
7836 i_bedict, be_new = self._GetUpdatedParams(
7837 instance.beparams, self.op.beparams,
7838 cluster.beparams[constants.PP_DEFAULT],
7839 constants.BES_PARAMETER_TYPES)
7840 self.be_new = be_new # the new actual values
7841 self.be_inst = i_bedict # the new dict (without defaults)
7843 self.be_new = self.be_inst = {}
7847 if constants.BE_MEMORY in self.op.beparams and not self.force:
7848 mem_check_list = [pnode]
7849 if be_new[constants.BE_AUTO_BALANCE]:
7850 # either we changed auto_balance to yes or it was from before
7851 mem_check_list.extend(instance.secondary_nodes)
7852 instance_info = self.rpc.call_instance_info(pnode, instance.name,
7853 instance.hypervisor)
7854 nodeinfo = self.rpc.call_node_info(mem_check_list, self.cfg.GetVGName(),
7855 instance.hypervisor)
7856 pninfo = nodeinfo[pnode]
7857 msg = pninfo.fail_msg
7859 # Assume the primary node is unreachable and go ahead
7860 self.warn.append("Can't get info from primary node %s: %s" %
7862 elif not isinstance(pninfo.payload.get('memory_free', None), int):
7863 self.warn.append("Node data from primary node %s doesn't contain"
7864 " free memory information" % pnode)
7865 elif instance_info.fail_msg:
7866 self.warn.append("Can't get instance runtime information: %s" %
7867 instance_info.fail_msg)
7869 if instance_info.payload:
7870 current_mem = int(instance_info.payload['memory'])
7872 # Assume instance not running
7873 # (there is a slight race condition here, but it's not very probable,
7874 # and we have no other way to check)
7876 miss_mem = (be_new[constants.BE_MEMORY] - current_mem -
7877 pninfo.payload['memory_free'])
7879 raise errors.OpPrereqError("This change will prevent the instance"
7880 " from starting, due to %d MB of memory"
7881 " missing on its primary node" % miss_mem,
7884 if be_new[constants.BE_AUTO_BALANCE]:
7885 for node, nres in nodeinfo.items():
7886 if node not in instance.secondary_nodes:
7890 self.warn.append("Can't get info from secondary node %s: %s" %
7892 elif not isinstance(nres.payload.get('memory_free', None), int):
7893 self.warn.append("Secondary node %s didn't return free"
7894 " memory information" % node)
7895 elif be_new[constants.BE_MEMORY] > nres.payload['memory_free']:
7896 self.warn.append("Not enough memory to failover instance to"
7897 " secondary node %s" % node)
7902 for nic_op, nic_dict in self.op.nics:
7903 if nic_op == constants.DDM_REMOVE:
7904 if not instance.nics:
7905 raise errors.OpPrereqError("Instance has no NICs, cannot remove",
7908 if nic_op != constants.DDM_ADD:
7910 if not instance.nics:
7911 raise errors.OpPrereqError("Invalid NIC index %s, instance has"
7912 " no NICs" % nic_op,
7914 if nic_op < 0 or nic_op >= len(instance.nics):
7915 raise errors.OpPrereqError("Invalid NIC index %s, valid values"
7917 (nic_op, len(instance.nics) - 1),
7919 old_nic_params = instance.nics[nic_op].nicparams
7920 old_nic_ip = instance.nics[nic_op].ip
7925 update_params_dict = dict([(key, nic_dict[key])
7926 for key in constants.NICS_PARAMETERS
7927 if key in nic_dict])
7929 if 'bridge' in nic_dict:
7930 update_params_dict[constants.NIC_LINK] = nic_dict['bridge']
7932 new_nic_params, new_filled_nic_params = \
7933 self._GetUpdatedParams(old_nic_params, update_params_dict,
7934 cluster.nicparams[constants.PP_DEFAULT],
7935 constants.NICS_PARAMETER_TYPES)
7936 objects.NIC.CheckParameterSyntax(new_filled_nic_params)
7937 self.nic_pinst[nic_op] = new_nic_params
7938 self.nic_pnew[nic_op] = new_filled_nic_params
7939 new_nic_mode = new_filled_nic_params[constants.NIC_MODE]
7941 if new_nic_mode == constants.NIC_MODE_BRIDGED:
7942 nic_bridge = new_filled_nic_params[constants.NIC_LINK]
7943 msg = self.rpc.call_bridges_exist(pnode, [nic_bridge]).fail_msg
7945 msg = "Error checking bridges on node %s: %s" % (pnode, msg)
7947 self.warn.append(msg)
7949 raise errors.OpPrereqError(msg, errors.ECODE_ENVIRON)
7950 if new_nic_mode == constants.NIC_MODE_ROUTED:
7951 if 'ip' in nic_dict:
7952 nic_ip = nic_dict['ip']
7956 raise errors.OpPrereqError('Cannot set the nic ip to None'
7957 ' on a routed nic', errors.ECODE_INVAL)
7958 if 'mac' in nic_dict:
7959 nic_mac = nic_dict['mac']
7961 raise errors.OpPrereqError('Cannot set the nic mac to None',
7963 elif nic_mac in (constants.VALUE_AUTO, constants.VALUE_GENERATE):
7964 # otherwise generate the mac
7965 nic_dict['mac'] = self.cfg.GenerateMAC(self.proc.GetECId())
7967 # or validate/reserve the current one
7969 self.cfg.ReserveMAC(nic_mac, self.proc.GetECId())
7970 except errors.ReservationError:
7971 raise errors.OpPrereqError("MAC address %s already in use"
7972 " in cluster" % nic_mac,
7973 errors.ECODE_NOTUNIQUE)
7976 if self.op.disks and instance.disk_template == constants.DT_DISKLESS:
7977 raise errors.OpPrereqError("Disk operations not supported for"
7978 " diskless instances",
7980 for disk_op, _ in self.op.disks:
7981 if disk_op == constants.DDM_REMOVE:
7982 if len(instance.disks) == 1:
7983 raise errors.OpPrereqError("Cannot remove the last disk of"
7986 ins_l = self.rpc.call_instance_list([pnode], [instance.hypervisor])
7987 ins_l = ins_l[pnode]
7988 msg = ins_l.fail_msg
7990 raise errors.OpPrereqError("Can't contact node %s: %s" %
7991 (pnode, msg), errors.ECODE_ENVIRON)
7992 if instance.name in ins_l.payload:
7993 raise errors.OpPrereqError("Instance is running, can't remove"
7994 " disks.", errors.ECODE_STATE)
7996 if (disk_op == constants.DDM_ADD and
7997 len(instance.nics) >= constants.MAX_DISKS):
7998 raise errors.OpPrereqError("Instance has too many disks (%d), cannot"
7999 " add more" % constants.MAX_DISKS,
8001 if disk_op not in (constants.DDM_ADD, constants.DDM_REMOVE):
8003 if disk_op < 0 or disk_op >= len(instance.disks):
8004 raise errors.OpPrereqError("Invalid disk index %s, valid values"
8006 (disk_op, len(instance.disks)),
8011 def Exec(self, feedback_fn):
8012 """Modifies an instance.
8014 All parameters take effect only at the next restart of the instance.
8017 # Process here the warnings from CheckPrereq, as we don't have a
8018 # feedback_fn there.
8019 for warn in self.warn:
8020 feedback_fn("WARNING: %s" % warn)
8023 instance = self.instance
8025 for disk_op, disk_dict in self.op.disks:
8026 if disk_op == constants.DDM_REMOVE:
8027 # remove the last disk
8028 device = instance.disks.pop()
8029 device_idx = len(instance.disks)
8030 for node, disk in device.ComputeNodeTree(instance.primary_node):
8031 self.cfg.SetDiskID(disk, node)
8032 msg = self.rpc.call_blockdev_remove(node, disk).fail_msg
8034 self.LogWarning("Could not remove disk/%d on node %s: %s,"
8035 " continuing anyway", device_idx, node, msg)
8036 result.append(("disk/%d" % device_idx, "remove"))
8037 elif disk_op == constants.DDM_ADD:
8039 if instance.disk_template == constants.DT_FILE:
8040 file_driver, file_path = instance.disks[0].logical_id
8041 file_path = os.path.dirname(file_path)
8043 file_driver = file_path = None
8044 disk_idx_base = len(instance.disks)
8045 new_disk = _GenerateDiskTemplate(self,
8046 instance.disk_template,
8047 instance.name, instance.primary_node,
8048 instance.secondary_nodes,
8053 instance.disks.append(new_disk)
8054 info = _GetInstanceInfoText(instance)
8056 logging.info("Creating volume %s for instance %s",
8057 new_disk.iv_name, instance.name)
8058 # Note: this needs to be kept in sync with _CreateDisks
8060 for node in instance.all_nodes:
8061 f_create = node == instance.primary_node
8063 _CreateBlockDev(self, node, instance, new_disk,
8064 f_create, info, f_create)
8065 except errors.OpExecError, err:
8066 self.LogWarning("Failed to create volume %s (%s) on"
8068 new_disk.iv_name, new_disk, node, err)
8069 result.append(("disk/%d" % disk_idx_base, "add:size=%s,mode=%s" %
8070 (new_disk.size, new_disk.mode)))
8072 # change a given disk
8073 instance.disks[disk_op].mode = disk_dict['mode']
8074 result.append(("disk.mode/%d" % disk_op, disk_dict['mode']))
8076 for nic_op, nic_dict in self.op.nics:
8077 if nic_op == constants.DDM_REMOVE:
8078 # remove the last nic
8079 del instance.nics[-1]
8080 result.append(("nic.%d" % len(instance.nics), "remove"))
8081 elif nic_op == constants.DDM_ADD:
8082 # mac and bridge should be set, by now
8083 mac = nic_dict['mac']
8084 ip = nic_dict.get('ip', None)
8085 nicparams = self.nic_pinst[constants.DDM_ADD]
8086 new_nic = objects.NIC(mac=mac, ip=ip, nicparams=nicparams)
8087 instance.nics.append(new_nic)
8088 result.append(("nic.%d" % (len(instance.nics) - 1),
8089 "add:mac=%s,ip=%s,mode=%s,link=%s" %
8090 (new_nic.mac, new_nic.ip,
8091 self.nic_pnew[constants.DDM_ADD][constants.NIC_MODE],
8092 self.nic_pnew[constants.DDM_ADD][constants.NIC_LINK]
8095 for key in 'mac', 'ip':
8097 setattr(instance.nics[nic_op], key, nic_dict[key])
8098 if nic_op in self.nic_pinst:
8099 instance.nics[nic_op].nicparams = self.nic_pinst[nic_op]
8100 for key, val in nic_dict.iteritems():
8101 result.append(("nic.%s/%d" % (key, nic_op), val))
8104 if self.op.hvparams:
8105 instance.hvparams = self.hv_inst
8106 for key, val in self.op.hvparams.iteritems():
8107 result.append(("hv/%s" % key, val))
8110 if self.op.beparams:
8111 instance.beparams = self.be_inst
8112 for key, val in self.op.beparams.iteritems():
8113 result.append(("be/%s" % key, val))
8115 self.cfg.Update(instance, feedback_fn)
8120 class LUQueryExports(NoHooksLU):
8121 """Query the exports list
8124 _OP_REQP = ['nodes']
8127 def ExpandNames(self):
8128 self.needed_locks = {}
8129 self.share_locks[locking.LEVEL_NODE] = 1
8130 if not self.op.nodes:
8131 self.needed_locks[locking.LEVEL_NODE] = locking.ALL_SET
8133 self.needed_locks[locking.LEVEL_NODE] = \
8134 _GetWantedNodes(self, self.op.nodes)
8136 def CheckPrereq(self):
8137 """Check prerequisites.
8140 self.nodes = self.acquired_locks[locking.LEVEL_NODE]
8142 def Exec(self, feedback_fn):
8143 """Compute the list of all the exported system images.
8146 @return: a dictionary with the structure node->(export-list)
8147 where export-list is a list of the instances exported on
8151 rpcresult = self.rpc.call_export_list(self.nodes)
8153 for node in rpcresult:
8154 if rpcresult[node].fail_msg:
8155 result[node] = False
8157 result[node] = rpcresult[node].payload
8162 class LUExportInstance(LogicalUnit):
8163 """Export an instance to an image in the cluster.
8166 HPATH = "instance-export"
8167 HTYPE = constants.HTYPE_INSTANCE
8168 _OP_REQP = ["instance_name", "target_node", "shutdown"]
8171 def CheckArguments(self):
8172 """Check the arguments.
8175 self.shutdown_timeout = getattr(self.op, "shutdown_timeout",
8176 constants.DEFAULT_SHUTDOWN_TIMEOUT)
8178 def ExpandNames(self):
8179 self._ExpandAndLockInstance()
8180 # FIXME: lock only instance primary and destination node
8182 # Sad but true, for now we have do lock all nodes, as we don't know where
8183 # the previous export might be, and and in this LU we search for it and
8184 # remove it from its current node. In the future we could fix this by:
8185 # - making a tasklet to search (share-lock all), then create the new one,
8186 # then one to remove, after
8187 # - removing the removal operation altogether
8188 self.needed_locks[locking.LEVEL_NODE] = locking.ALL_SET
8190 def DeclareLocks(self, level):
8191 """Last minute lock declaration."""
8192 # All nodes are locked anyway, so nothing to do here.
8194 def BuildHooksEnv(self):
8197 This will run on the master, primary node and target node.
8201 "EXPORT_NODE": self.op.target_node,
8202 "EXPORT_DO_SHUTDOWN": self.op.shutdown,
8203 "SHUTDOWN_TIMEOUT": self.shutdown_timeout,
8205 env.update(_BuildInstanceHookEnvByObject(self, self.instance))
8206 nl = [self.cfg.GetMasterNode(), self.instance.primary_node,
8207 self.op.target_node]
8210 def CheckPrereq(self):
8211 """Check prerequisites.
8213 This checks that the instance and node names are valid.
8216 instance_name = self.op.instance_name
8217 self.instance = self.cfg.GetInstanceInfo(instance_name)
8218 assert self.instance is not None, \
8219 "Cannot retrieve locked instance %s" % self.op.instance_name
8220 _CheckNodeOnline(self, self.instance.primary_node)
8222 self.op.target_node = _ExpandNodeName(self.cfg, self.op.target_node)
8223 self.dst_node = self.cfg.GetNodeInfo(self.op.target_node)
8224 assert self.dst_node is not None
8226 _CheckNodeOnline(self, self.dst_node.name)
8227 _CheckNodeNotDrained(self, self.dst_node.name)
8229 # instance disk type verification
8230 for disk in self.instance.disks:
8231 if disk.dev_type == constants.LD_FILE:
8232 raise errors.OpPrereqError("Export not supported for instances with"
8233 " file-based disks", errors.ECODE_INVAL)
8235 def Exec(self, feedback_fn):
8236 """Export an instance to an image in the cluster.
8239 instance = self.instance
8240 dst_node = self.dst_node
8241 src_node = instance.primary_node
8243 if self.op.shutdown:
8244 # shutdown the instance, but not the disks
8245 feedback_fn("Shutting down instance %s" % instance.name)
8246 result = self.rpc.call_instance_shutdown(src_node, instance,
8247 self.shutdown_timeout)
8248 result.Raise("Could not shutdown instance %s on"
8249 " node %s" % (instance.name, src_node))
8251 vgname = self.cfg.GetVGName()
8255 # set the disks ID correctly since call_instance_start needs the
8256 # correct drbd minor to create the symlinks
8257 for disk in instance.disks:
8258 self.cfg.SetDiskID(disk, src_node)
8260 activate_disks = (not instance.admin_up)
8263 # Activate the instance disks if we'exporting a stopped instance
8264 feedback_fn("Activating disks for %s" % instance.name)
8265 _StartInstanceDisks(self, instance, None)
8271 for idx, disk in enumerate(instance.disks):
8272 feedback_fn("Creating a snapshot of disk/%s on node %s" %
8275 # result.payload will be a snapshot of an lvm leaf of the one we
8277 result = self.rpc.call_blockdev_snapshot(src_node, disk)
8278 msg = result.fail_msg
8280 self.LogWarning("Could not snapshot disk/%s on node %s: %s",
8282 snap_disks.append(False)
8284 disk_id = (vgname, result.payload)
8285 new_dev = objects.Disk(dev_type=constants.LD_LV, size=disk.size,
8286 logical_id=disk_id, physical_id=disk_id,
8287 iv_name=disk.iv_name)
8288 snap_disks.append(new_dev)
8291 if self.op.shutdown and instance.admin_up:
8292 feedback_fn("Starting instance %s" % instance.name)
8293 result = self.rpc.call_instance_start(src_node, instance, None, None)
8294 msg = result.fail_msg
8296 _ShutdownInstanceDisks(self, instance)
8297 raise errors.OpExecError("Could not start instance: %s" % msg)
8299 # TODO: check for size
8301 cluster_name = self.cfg.GetClusterName()
8302 for idx, dev in enumerate(snap_disks):
8303 feedback_fn("Exporting snapshot %s from %s to %s" %
8304 (idx, src_node, dst_node.name))
8306 # FIXME: pass debug from opcode to backend
8307 result = self.rpc.call_snapshot_export(src_node, dev, dst_node.name,
8308 instance, cluster_name,
8309 idx, self.op.debug_level)
8310 msg = result.fail_msg
8312 self.LogWarning("Could not export disk/%s from node %s to"
8313 " node %s: %s", idx, src_node, dst_node.name, msg)
8314 dresults.append(False)
8316 dresults.append(True)
8317 msg = self.rpc.call_blockdev_remove(src_node, dev).fail_msg
8319 self.LogWarning("Could not remove snapshot for disk/%d from node"
8320 " %s: %s", idx, src_node, msg)
8322 dresults.append(False)
8324 feedback_fn("Finalizing export on %s" % dst_node.name)
8325 result = self.rpc.call_finalize_export(dst_node.name, instance,
8328 msg = result.fail_msg
8330 self.LogWarning("Could not finalize export for instance %s"
8331 " on node %s: %s", instance.name, dst_node.name, msg)
8336 feedback_fn("Deactivating disks for %s" % instance.name)
8337 _ShutdownInstanceDisks(self, instance)
8339 nodelist = self.cfg.GetNodeList()
8340 nodelist.remove(dst_node.name)
8342 # on one-node clusters nodelist will be empty after the removal
8343 # if we proceed the backup would be removed because OpQueryExports
8344 # substitutes an empty list with the full cluster node list.
8345 iname = instance.name
8347 feedback_fn("Removing old exports for instance %s" % iname)
8348 exportlist = self.rpc.call_export_list(nodelist)
8349 for node in exportlist:
8350 if exportlist[node].fail_msg:
8352 if iname in exportlist[node].payload:
8353 msg = self.rpc.call_export_remove(node, iname).fail_msg
8355 self.LogWarning("Could not remove older export for instance %s"
8356 " on node %s: %s", iname, node, msg)
8357 return fin_resu, dresults
8360 class LURemoveExport(NoHooksLU):
8361 """Remove exports related to the named instance.
8364 _OP_REQP = ["instance_name"]
8367 def ExpandNames(self):
8368 self.needed_locks = {}
8369 # We need all nodes to be locked in order for RemoveExport to work, but we
8370 # don't need to lock the instance itself, as nothing will happen to it (and
8371 # we can remove exports also for a removed instance)
8372 self.needed_locks[locking.LEVEL_NODE] = locking.ALL_SET
8374 def CheckPrereq(self):
8375 """Check prerequisites.
8379 def Exec(self, feedback_fn):
8380 """Remove any export.
8383 instance_name = self.cfg.ExpandInstanceName(self.op.instance_name)
8384 # If the instance was not found we'll try with the name that was passed in.
8385 # This will only work if it was an FQDN, though.
8387 if not instance_name:
8389 instance_name = self.op.instance_name
8391 locked_nodes = self.acquired_locks[locking.LEVEL_NODE]
8392 exportlist = self.rpc.call_export_list(locked_nodes)
8394 for node in exportlist:
8395 msg = exportlist[node].fail_msg
8397 self.LogWarning("Failed to query node %s (continuing): %s", node, msg)
8399 if instance_name in exportlist[node].payload:
8401 result = self.rpc.call_export_remove(node, instance_name)
8402 msg = result.fail_msg
8404 logging.error("Could not remove export for instance %s"
8405 " on node %s: %s", instance_name, node, msg)
8407 if fqdn_warn and not found:
8408 feedback_fn("Export not found. If trying to remove an export belonging"
8409 " to a deleted instance please use its Fully Qualified"
8413 class TagsLU(NoHooksLU): # pylint: disable-msg=W0223
8416 This is an abstract class which is the parent of all the other tags LUs.
8420 def ExpandNames(self):
8421 self.needed_locks = {}
8422 if self.op.kind == constants.TAG_NODE:
8423 self.op.name = _ExpandNodeName(self.cfg, self.op.name)
8424 self.needed_locks[locking.LEVEL_NODE] = self.op.name
8425 elif self.op.kind == constants.TAG_INSTANCE:
8426 self.op.name = _ExpandInstanceName(self.cfg, self.op.name)
8427 self.needed_locks[locking.LEVEL_INSTANCE] = self.op.name
8429 def CheckPrereq(self):
8430 """Check prerequisites.
8433 if self.op.kind == constants.TAG_CLUSTER:
8434 self.target = self.cfg.GetClusterInfo()
8435 elif self.op.kind == constants.TAG_NODE:
8436 self.target = self.cfg.GetNodeInfo(self.op.name)
8437 elif self.op.kind == constants.TAG_INSTANCE:
8438 self.target = self.cfg.GetInstanceInfo(self.op.name)
8440 raise errors.OpPrereqError("Wrong tag type requested (%s)" %
8441 str(self.op.kind), errors.ECODE_INVAL)
8444 class LUGetTags(TagsLU):
8445 """Returns the tags of a given object.
8448 _OP_REQP = ["kind", "name"]
8451 def Exec(self, feedback_fn):
8452 """Returns the tag list.
8455 return list(self.target.GetTags())
8458 class LUSearchTags(NoHooksLU):
8459 """Searches the tags for a given pattern.
8462 _OP_REQP = ["pattern"]
8465 def ExpandNames(self):
8466 self.needed_locks = {}
8468 def CheckPrereq(self):
8469 """Check prerequisites.
8471 This checks the pattern passed for validity by compiling it.
8475 self.re = re.compile(self.op.pattern)
8476 except re.error, err:
8477 raise errors.OpPrereqError("Invalid search pattern '%s': %s" %
8478 (self.op.pattern, err), errors.ECODE_INVAL)
8480 def Exec(self, feedback_fn):
8481 """Returns the tag list.
8485 tgts = [("/cluster", cfg.GetClusterInfo())]
8486 ilist = cfg.GetAllInstancesInfo().values()
8487 tgts.extend([("/instances/%s" % i.name, i) for i in ilist])
8488 nlist = cfg.GetAllNodesInfo().values()
8489 tgts.extend([("/nodes/%s" % n.name, n) for n in nlist])
8491 for path, target in tgts:
8492 for tag in target.GetTags():
8493 if self.re.search(tag):
8494 results.append((path, tag))
8498 class LUAddTags(TagsLU):
8499 """Sets a tag on a given object.
8502 _OP_REQP = ["kind", "name", "tags"]
8505 def CheckPrereq(self):
8506 """Check prerequisites.
8508 This checks the type and length of the tag name and value.
8511 TagsLU.CheckPrereq(self)
8512 for tag in self.op.tags:
8513 objects.TaggableObject.ValidateTag(tag)
8515 def Exec(self, feedback_fn):
8520 for tag in self.op.tags:
8521 self.target.AddTag(tag)
8522 except errors.TagError, err:
8523 raise errors.OpExecError("Error while setting tag: %s" % str(err))
8524 self.cfg.Update(self.target, feedback_fn)
8527 class LUDelTags(TagsLU):
8528 """Delete a list of tags from a given object.
8531 _OP_REQP = ["kind", "name", "tags"]
8534 def CheckPrereq(self):
8535 """Check prerequisites.
8537 This checks that we have the given tag.
8540 TagsLU.CheckPrereq(self)
8541 for tag in self.op.tags:
8542 objects.TaggableObject.ValidateTag(tag)
8543 del_tags = frozenset(self.op.tags)
8544 cur_tags = self.target.GetTags()
8545 if not del_tags <= cur_tags:
8546 diff_tags = del_tags - cur_tags
8547 diff_names = ["'%s'" % tag for tag in diff_tags]
8549 raise errors.OpPrereqError("Tag(s) %s not found" %
8550 (",".join(diff_names)), errors.ECODE_NOENT)
8552 def Exec(self, feedback_fn):
8553 """Remove the tag from the object.
8556 for tag in self.op.tags:
8557 self.target.RemoveTag(tag)
8558 self.cfg.Update(self.target, feedback_fn)
8561 class LUTestDelay(NoHooksLU):
8562 """Sleep for a specified amount of time.
8564 This LU sleeps on the master and/or nodes for a specified amount of
8568 _OP_REQP = ["duration", "on_master", "on_nodes"]
8571 def ExpandNames(self):
8572 """Expand names and set required locks.
8574 This expands the node list, if any.
8577 self.needed_locks = {}
8578 if self.op.on_nodes:
8579 # _GetWantedNodes can be used here, but is not always appropriate to use
8580 # this way in ExpandNames. Check LogicalUnit.ExpandNames docstring for
8582 self.op.on_nodes = _GetWantedNodes(self, self.op.on_nodes)
8583 self.needed_locks[locking.LEVEL_NODE] = self.op.on_nodes
8585 def CheckPrereq(self):
8586 """Check prerequisites.
8590 def Exec(self, feedback_fn):
8591 """Do the actual sleep.
8594 if self.op.on_master:
8595 if not utils.TestDelay(self.op.duration):
8596 raise errors.OpExecError("Error during master delay test")
8597 if self.op.on_nodes:
8598 result = self.rpc.call_test_delay(self.op.on_nodes, self.op.duration)
8599 for node, node_result in result.items():
8600 node_result.Raise("Failure during rpc call to node %s" % node)
8603 class IAllocator(object):
8604 """IAllocator framework.
8606 An IAllocator instance has three sets of attributes:
8607 - cfg that is needed to query the cluster
8608 - input data (all members of the _KEYS class attribute are required)
8609 - four buffer attributes (in|out_data|text), that represent the
8610 input (to the external script) in text and data structure format,
8611 and the output from it, again in two formats
8612 - the result variables from the script (success, info, nodes) for
8616 # pylint: disable-msg=R0902
8617 # lots of instance attributes
8619 "name", "mem_size", "disks", "disk_template",
8620 "os", "tags", "nics", "vcpus", "hypervisor",
8623 "name", "relocate_from",
8629 def __init__(self, cfg, rpc, mode, **kwargs):
8632 # init buffer variables
8633 self.in_text = self.out_text = self.in_data = self.out_data = None
8634 # init all input fields so that pylint is happy
8636 self.mem_size = self.disks = self.disk_template = None
8637 self.os = self.tags = self.nics = self.vcpus = None
8638 self.hypervisor = None
8639 self.relocate_from = None
8641 self.evac_nodes = None
8643 self.required_nodes = None
8644 # init result fields
8645 self.success = self.info = self.result = None
8646 if self.mode == constants.IALLOCATOR_MODE_ALLOC:
8647 keyset = self._ALLO_KEYS
8648 fn = self._AddNewInstance
8649 elif self.mode == constants.IALLOCATOR_MODE_RELOC:
8650 keyset = self._RELO_KEYS
8651 fn = self._AddRelocateInstance
8652 elif self.mode == constants.IALLOCATOR_MODE_MEVAC:
8653 keyset = self._EVAC_KEYS
8654 fn = self._AddEvacuateNodes
8656 raise errors.ProgrammerError("Unknown mode '%s' passed to the"
8657 " IAllocator" % self.mode)
8659 if key not in keyset:
8660 raise errors.ProgrammerError("Invalid input parameter '%s' to"
8661 " IAllocator" % key)
8662 setattr(self, key, kwargs[key])
8665 if key not in kwargs:
8666 raise errors.ProgrammerError("Missing input parameter '%s' to"
8667 " IAllocator" % key)
8668 self._BuildInputData(fn)
8670 def _ComputeClusterData(self):
8671 """Compute the generic allocator input data.
8673 This is the data that is independent of the actual operation.
8677 cluster_info = cfg.GetClusterInfo()
8680 "version": constants.IALLOCATOR_VERSION,
8681 "cluster_name": cfg.GetClusterName(),
8682 "cluster_tags": list(cluster_info.GetTags()),
8683 "enabled_hypervisors": list(cluster_info.enabled_hypervisors),
8684 # we don't have job IDs
8686 iinfo = cfg.GetAllInstancesInfo().values()
8687 i_list = [(inst, cluster_info.FillBE(inst)) for inst in iinfo]
8691 node_list = cfg.GetNodeList()
8693 if self.mode == constants.IALLOCATOR_MODE_ALLOC:
8694 hypervisor_name = self.hypervisor
8695 elif self.mode == constants.IALLOCATOR_MODE_RELOC:
8696 hypervisor_name = cfg.GetInstanceInfo(self.name).hypervisor
8697 elif self.mode == constants.IALLOCATOR_MODE_MEVAC:
8698 hypervisor_name = cluster_info.enabled_hypervisors[0]
8700 node_data = self.rpc.call_node_info(node_list, cfg.GetVGName(),
8703 self.rpc.call_all_instances_info(node_list,
8704 cluster_info.enabled_hypervisors)
8705 for nname, nresult in node_data.items():
8706 # first fill in static (config-based) values
8707 ninfo = cfg.GetNodeInfo(nname)
8709 "tags": list(ninfo.GetTags()),
8710 "primary_ip": ninfo.primary_ip,
8711 "secondary_ip": ninfo.secondary_ip,
8712 "offline": ninfo.offline,
8713 "drained": ninfo.drained,
8714 "master_candidate": ninfo.master_candidate,
8717 if not (ninfo.offline or ninfo.drained):
8718 nresult.Raise("Can't get data for node %s" % nname)
8719 node_iinfo[nname].Raise("Can't get node instance info from node %s" %
8721 remote_info = nresult.payload
8723 for attr in ['memory_total', 'memory_free', 'memory_dom0',
8724 'vg_size', 'vg_free', 'cpu_total']:
8725 if attr not in remote_info:
8726 raise errors.OpExecError("Node '%s' didn't return attribute"
8727 " '%s'" % (nname, attr))
8728 if not isinstance(remote_info[attr], int):
8729 raise errors.OpExecError("Node '%s' returned invalid value"
8731 (nname, attr, remote_info[attr]))
8732 # compute memory used by primary instances
8733 i_p_mem = i_p_up_mem = 0
8734 for iinfo, beinfo in i_list:
8735 if iinfo.primary_node == nname:
8736 i_p_mem += beinfo[constants.BE_MEMORY]
8737 if iinfo.name not in node_iinfo[nname].payload:
8740 i_used_mem = int(node_iinfo[nname].payload[iinfo.name]['memory'])
8741 i_mem_diff = beinfo[constants.BE_MEMORY] - i_used_mem
8742 remote_info['memory_free'] -= max(0, i_mem_diff)
8745 i_p_up_mem += beinfo[constants.BE_MEMORY]
8747 # compute memory used by instances
8749 "total_memory": remote_info['memory_total'],
8750 "reserved_memory": remote_info['memory_dom0'],
8751 "free_memory": remote_info['memory_free'],
8752 "total_disk": remote_info['vg_size'],
8753 "free_disk": remote_info['vg_free'],
8754 "total_cpus": remote_info['cpu_total'],
8755 "i_pri_memory": i_p_mem,
8756 "i_pri_up_memory": i_p_up_mem,
8760 node_results[nname] = pnr
8761 data["nodes"] = node_results
8765 for iinfo, beinfo in i_list:
8767 for nic in iinfo.nics:
8768 filled_params = objects.FillDict(
8769 cluster_info.nicparams[constants.PP_DEFAULT],
8771 nic_dict = {"mac": nic.mac,
8773 "mode": filled_params[constants.NIC_MODE],
8774 "link": filled_params[constants.NIC_LINK],
8776 if filled_params[constants.NIC_MODE] == constants.NIC_MODE_BRIDGED:
8777 nic_dict["bridge"] = filled_params[constants.NIC_LINK]
8778 nic_data.append(nic_dict)
8780 "tags": list(iinfo.GetTags()),
8781 "admin_up": iinfo.admin_up,
8782 "vcpus": beinfo[constants.BE_VCPUS],
8783 "memory": beinfo[constants.BE_MEMORY],
8785 "nodes": [iinfo.primary_node] + list(iinfo.secondary_nodes),
8787 "disks": [{"size": dsk.size, "mode": dsk.mode} for dsk in iinfo.disks],
8788 "disk_template": iinfo.disk_template,
8789 "hypervisor": iinfo.hypervisor,
8791 pir["disk_space_total"] = _ComputeDiskSize(iinfo.disk_template,
8793 instance_data[iinfo.name] = pir
8795 data["instances"] = instance_data
8799 def _AddNewInstance(self):
8800 """Add new instance data to allocator structure.
8802 This in combination with _AllocatorGetClusterData will create the
8803 correct structure needed as input for the allocator.
8805 The checks for the completeness of the opcode must have already been
8809 disk_space = _ComputeDiskSize(self.disk_template, self.disks)
8811 if self.disk_template in constants.DTS_NET_MIRROR:
8812 self.required_nodes = 2
8814 self.required_nodes = 1
8817 "disk_template": self.disk_template,
8820 "vcpus": self.vcpus,
8821 "memory": self.mem_size,
8822 "disks": self.disks,
8823 "disk_space_total": disk_space,
8825 "required_nodes": self.required_nodes,
8829 def _AddRelocateInstance(self):
8830 """Add relocate instance data to allocator structure.
8832 This in combination with _IAllocatorGetClusterData will create the
8833 correct structure needed as input for the allocator.
8835 The checks for the completeness of the opcode must have already been
8839 instance = self.cfg.GetInstanceInfo(self.name)
8840 if instance is None:
8841 raise errors.ProgrammerError("Unknown instance '%s' passed to"
8842 " IAllocator" % self.name)
8844 if instance.disk_template not in constants.DTS_NET_MIRROR:
8845 raise errors.OpPrereqError("Can't relocate non-mirrored instances",
8848 if len(instance.secondary_nodes) != 1:
8849 raise errors.OpPrereqError("Instance has not exactly one secondary node",
8852 self.required_nodes = 1
8853 disk_sizes = [{'size': disk.size} for disk in instance.disks]
8854 disk_space = _ComputeDiskSize(instance.disk_template, disk_sizes)
8858 "disk_space_total": disk_space,
8859 "required_nodes": self.required_nodes,
8860 "relocate_from": self.relocate_from,
8864 def _AddEvacuateNodes(self):
8865 """Add evacuate nodes data to allocator structure.
8869 "evac_nodes": self.evac_nodes
8873 def _BuildInputData(self, fn):
8874 """Build input data structures.
8877 self._ComputeClusterData()
8880 request["type"] = self.mode
8881 self.in_data["request"] = request
8883 self.in_text = serializer.Dump(self.in_data)
8885 def Run(self, name, validate=True, call_fn=None):
8886 """Run an instance allocator and return the results.
8890 call_fn = self.rpc.call_iallocator_runner
8892 result = call_fn(self.cfg.GetMasterNode(), name, self.in_text)
8893 result.Raise("Failure while running the iallocator script")
8895 self.out_text = result.payload
8897 self._ValidateResult()
8899 def _ValidateResult(self):
8900 """Process the allocator results.
8902 This will process and if successful save the result in
8903 self.out_data and the other parameters.
8907 rdict = serializer.Load(self.out_text)
8908 except Exception, err:
8909 raise errors.OpExecError("Can't parse iallocator results: %s" % str(err))
8911 if not isinstance(rdict, dict):
8912 raise errors.OpExecError("Can't parse iallocator results: not a dict")
8914 # TODO: remove backwards compatiblity in later versions
8915 if "nodes" in rdict and "result" not in rdict:
8916 rdict["result"] = rdict["nodes"]
8919 for key in "success", "info", "result":
8920 if key not in rdict:
8921 raise errors.OpExecError("Can't parse iallocator results:"
8922 " missing key '%s'" % key)
8923 setattr(self, key, rdict[key])
8925 if not isinstance(rdict["result"], list):
8926 raise errors.OpExecError("Can't parse iallocator results: 'result' key"
8928 self.out_data = rdict
8931 class LUTestAllocator(NoHooksLU):
8932 """Run allocator tests.
8934 This LU runs the allocator tests
8937 _OP_REQP = ["direction", "mode", "name"]
8939 def CheckPrereq(self):
8940 """Check prerequisites.
8942 This checks the opcode parameters depending on the director and mode test.
8945 if self.op.mode == constants.IALLOCATOR_MODE_ALLOC:
8946 for attr in ["name", "mem_size", "disks", "disk_template",
8947 "os", "tags", "nics", "vcpus"]:
8948 if not hasattr(self.op, attr):
8949 raise errors.OpPrereqError("Missing attribute '%s' on opcode input" %
8950 attr, errors.ECODE_INVAL)
8951 iname = self.cfg.ExpandInstanceName(self.op.name)
8952 if iname is not None:
8953 raise errors.OpPrereqError("Instance '%s' already in the cluster" %
8954 iname, errors.ECODE_EXISTS)
8955 if not isinstance(self.op.nics, list):
8956 raise errors.OpPrereqError("Invalid parameter 'nics'",
8958 for row in self.op.nics:
8959 if (not isinstance(row, dict) or
8962 "bridge" not in row):
8963 raise errors.OpPrereqError("Invalid contents of the 'nics'"
8964 " parameter", errors.ECODE_INVAL)
8965 if not isinstance(self.op.disks, list):
8966 raise errors.OpPrereqError("Invalid parameter 'disks'",
8968 for row in self.op.disks:
8969 if (not isinstance(row, dict) or
8970 "size" not in row or
8971 not isinstance(row["size"], int) or
8972 "mode" not in row or
8973 row["mode"] not in ['r', 'w']):
8974 raise errors.OpPrereqError("Invalid contents of the 'disks'"
8975 " parameter", errors.ECODE_INVAL)
8976 if not hasattr(self.op, "hypervisor") or self.op.hypervisor is None:
8977 self.op.hypervisor = self.cfg.GetHypervisorType()
8978 elif self.op.mode == constants.IALLOCATOR_MODE_RELOC:
8979 if not hasattr(self.op, "name"):
8980 raise errors.OpPrereqError("Missing attribute 'name' on opcode input",
8982 fname = _ExpandInstanceName(self.cfg, self.op.name)
8983 self.op.name = fname
8984 self.relocate_from = self.cfg.GetInstanceInfo(fname).secondary_nodes
8985 elif self.op.mode == constants.IALLOCATOR_MODE_MEVAC:
8986 if not hasattr(self.op, "evac_nodes"):
8987 raise errors.OpPrereqError("Missing attribute 'evac_nodes' on"
8988 " opcode input", errors.ECODE_INVAL)
8990 raise errors.OpPrereqError("Invalid test allocator mode '%s'" %
8991 self.op.mode, errors.ECODE_INVAL)
8993 if self.op.direction == constants.IALLOCATOR_DIR_OUT:
8994 if not hasattr(self.op, "allocator") or self.op.allocator is None:
8995 raise errors.OpPrereqError("Missing allocator name",
8997 elif self.op.direction != constants.IALLOCATOR_DIR_IN:
8998 raise errors.OpPrereqError("Wrong allocator test '%s'" %
8999 self.op.direction, errors.ECODE_INVAL)
9001 def Exec(self, feedback_fn):
9002 """Run the allocator test.
9005 if self.op.mode == constants.IALLOCATOR_MODE_ALLOC:
9006 ial = IAllocator(self.cfg, self.rpc,
9009 mem_size=self.op.mem_size,
9010 disks=self.op.disks,
9011 disk_template=self.op.disk_template,
9015 vcpus=self.op.vcpus,
9016 hypervisor=self.op.hypervisor,
9018 elif self.op.mode == constants.IALLOCATOR_MODE_RELOC:
9019 ial = IAllocator(self.cfg, self.rpc,
9022 relocate_from=list(self.relocate_from),
9024 elif self.op.mode == constants.IALLOCATOR_MODE_MEVAC:
9025 ial = IAllocator(self.cfg, self.rpc,
9027 evac_nodes=self.op.evac_nodes)
9029 raise errors.ProgrammerError("Uncatched mode %s in"
9030 " LUTestAllocator.Exec", self.op.mode)
9032 if self.op.direction == constants.IALLOCATOR_DIR_IN:
9033 result = ial.in_text
9035 ial.Run(self.op.allocator, validate=False)
9036 result = ial.out_text