1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31 """Module implementing the iallocator code."""
32
33 from ganeti import compat
34 from ganeti import constants
35 from ganeti import errors
36 from ganeti import ht
37 from ganeti import outils
38 from ganeti import opcodes
39 import ganeti.rpc.node as rpc
40 from ganeti import serializer
41 from ganeti import utils
42
43 import ganeti.masterd.instance as gmi
44
45 import logging
46
47 _STRING_LIST = ht.TListOf(ht.TString)
48 _JOB_LIST = ht.TListOf(ht.TListOf(ht.TStrictDict(True, False, {
49
50
51 "OP_ID": ht.TElemOf([opcodes.OpInstanceFailover.OP_ID,
52 opcodes.OpInstanceMigrate.OP_ID,
53 opcodes.OpInstanceReplaceDisks.OP_ID]),
54 })))
55
56 _NEVAC_MOVED = \
57 ht.TListOf(ht.TAnd(ht.TIsLength(3),
58 ht.TItems([ht.TNonEmptyString,
59 ht.TNonEmptyString,
60 ht.TListOf(ht.TNonEmptyString),
61 ])))
62 _NEVAC_FAILED = \
63 ht.TListOf(ht.TAnd(ht.TIsLength(2),
64 ht.TItems([ht.TNonEmptyString,
65 ht.TMaybeString,
66 ])))
67 _NEVAC_RESULT = ht.TAnd(ht.TIsLength(3),
68 ht.TItems([_NEVAC_MOVED, _NEVAC_FAILED, _JOB_LIST]))
69
70 _INST_NAME = ("name", ht.TNonEmptyString)
71 _INST_UUID = ("inst_uuid", ht.TNonEmptyString)
75 """Meta class for request definitions.
76
77 """
78 @classmethod
80 """Extract the slots out of REQ_PARAMS.
81
82 """
83 params = attrs.setdefault("REQ_PARAMS", [])
84 return [slot for (slot, _) in params]
85
88 """A generic IAllocator request object.
89
90 """
91 __metaclass__ = _AutoReqParam
92
93 MODE = NotImplemented
94 REQ_PARAMS = []
95 REQ_RESULT = NotImplemented
96
98 """Constructor for IARequestBase.
99
100 The constructor takes only keyword arguments and will set
101 attributes on this object based on the passed arguments. As such,
102 it means that you should not pass arguments which are not in the
103 REQ_PARAMS attribute for this class.
104
105 """
106 outils.ValidatedSlots.__init__(self, **kwargs)
107
108 self.Validate()
109
111 """Validates all parameters of the request.
112
113
114 This method returns L{None} if the validation succeeds, or raises
115 an exception otherwise.
116
117 @rtype: NoneType
118 @return: L{None}, if the validation succeeds
119
120 @raise Exception: validation fails
121
122 """
123 assert self.MODE in constants.VALID_IALLOCATOR_MODES
124
125 for (param, validator) in self.REQ_PARAMS:
126 if not hasattr(self, param):
127 raise errors.OpPrereqError("Request is missing '%s' parameter" % param,
128 errors.ECODE_INVAL)
129
130 value = getattr(self, param)
131 if not validator(value):
132 raise errors.OpPrereqError(("Request parameter '%s' has invalid"
133 " type %s/value %s") %
134 (param, type(value), value),
135 errors.ECODE_INVAL)
136
138 """Gets the request data dict.
139
140 @param cfg: The configuration instance
141
142 """
143 raise NotImplementedError
144
146 """Gets extra parameters to the IAllocator call.
147
148 """
149 return {}
150
152 """Validates the result of an request.
153
154 @param ia: The IAllocator instance
155 @param result: The IAllocator run result
156 @raises ResultValidationError: If validation fails
157
158 """
159 if ia.success and not self.REQ_RESULT(result):
160 raise errors.ResultValidationError("iallocator returned invalid result,"
161 " expected %s, got %s" %
162 (self.REQ_RESULT, result))
163
166 """An instance allocation request.
167
168 """
169
170 MODE = constants.IALLOCATOR_MODE_ALLOC
171 REQ_PARAMS = [
172 _INST_NAME,
173 ("memory", ht.TNonNegativeInt),
174 ("spindle_use", ht.TNonNegativeInt),
175 ("disks", ht.TListOf(ht.TDict)),
176 ("disk_template", ht.TString),
177 ("group_name", ht.TMaybe(ht.TNonEmptyString)),
178 ("os", ht.TString),
179 ("tags", _STRING_LIST),
180 ("nics", ht.TListOf(ht.TDict)),
181 ("vcpus", ht.TInt),
182 ("hypervisor", ht.TString),
183 ("node_whitelist", ht.TMaybeListOf(ht.TNonEmptyString)),
184 ]
185 REQ_RESULT = ht.TList
186
188 """Calculates the required nodes based on the disk_template.
189
190 """
191 if self.disk_template in constants.DTS_INT_MIRROR:
192 return 2
193 else:
194 return 1
195
197 """Requests a new instance.
198
199 The checks for the completeness of the opcode must have already been
200 done.
201
202 """
203 for d in self.disks:
204 d[constants.IDISK_TYPE] = self.disk_template
205 disk_space = gmi.ComputeDiskSize(self.disks)
206
207 return {
208 "name": self.name,
209 "disk_template": self.disk_template,
210 "group_name": self.group_name,
211 "tags": self.tags,
212 "os": self.os,
213 "vcpus": self.vcpus,
214 "memory": self.memory,
215 "spindle_use": self.spindle_use,
216 "disks": self.disks,
217 "disk_space_total": disk_space,
218 "nics": self.nics,
219 "required_nodes": self.RequiredNodes(),
220 "hypervisor": self.hypervisor,
221 }
222
233
250
274
277 """A relocation request.
278
279 """
280
281 MODE = constants.IALLOCATOR_MODE_RELOC
282 REQ_PARAMS = [
283 _INST_UUID,
284 ("relocate_from_node_uuids", _STRING_LIST),
285 ]
286 REQ_RESULT = ht.TList
287
321
345
346 @staticmethod
348 """Returns a list of unique group names for a list of nodes.
349
350 @type node2group: dict
351 @param node2group: Map from node name to group UUID
352 @type groups: dict
353 @param groups: Group information
354 @type nodes: list
355 @param nodes: Node names
356
357 """
358 result = set()
359
360 for node in nodes:
361 try:
362 group_uuid = node2group[node]
363 except KeyError:
364
365 pass
366 else:
367 try:
368 group = groups[group_uuid]
369 except KeyError:
370
371 group_name = group_uuid
372 else:
373 group_name = group["name"]
374
375 result.add(group_name)
376
377 return sorted(result)
378
381 """A node evacuation request.
382
383 """
384
385 MODE = constants.IALLOCATOR_MODE_NODE_EVAC
386 REQ_PARAMS = [
387 ("instances", _STRING_LIST),
388 ("evac_mode", ht.TEvacMode),
389 ("ignore_soft_errors", ht.TMaybe(ht.TBool)),
390 ]
391 REQ_RESULT = _NEVAC_RESULT
392
394 """Get data for node-evacuate requests.
395
396 """
397 return {
398 "instances": self.instances,
399 "evac_mode": self.evac_mode,
400 }
401
403 """Get extra iallocator command line options for
404 node-evacuate requests.
405
406 """
407 if self.ignore_soft_errors:
408 return {"ignore-soft-errors": None}
409 else:
410 return {}
411
433
436 """IAllocator framework.
437
438 An IAllocator instance has three sets of attributes:
439 - cfg that is needed to query the cluster
440 - input data (all members of the _KEYS class attribute are required)
441 - four buffer attributes (in|out_data|text), that represent the
442 input (to the external script) in text and data structure format,
443 and the output from it, again in two formats
444 - the result variables from the script (success, info, nodes) for
445 easy usage
446
447 """
448
449
450
451 - def __init__(self, cfg, rpc_runner, req):
452 self.cfg = cfg
453 self.rpc = rpc_runner
454 self.req = req
455
456 self.in_text = self.out_text = self.in_data = self.out_data = None
457
458 self.success = self.info = self.result = None
459
460 self._BuildInputData(req)
461
464 """Prepare and execute node info call.
465
466 @type disk_templates: list of string
467 @param disk_templates: the disk templates of the instances to be allocated
468 @type node_list: list of strings
469 @param node_list: list of nodes' UUIDs
470 @type cluster_info: L{objects.Cluster}
471 @param cluster_info: the cluster's information from the config
472 @type hypervisor_name: string
473 @param hypervisor_name: the hypervisor name
474 @rtype: same as the result of the node info RPC call
475 @return: the result of the node info RPC call
476
477 """
478 storage_units_raw = utils.storage.GetStorageUnits(self.cfg, disk_templates)
479 storage_units = rpc.PrepareStorageUnitsForNodes(self.cfg, storage_units_raw,
480 node_list)
481 hvspecs = [(hypervisor_name, cluster_info.hvparams[hypervisor_name])]
482 return self.rpc.call_node_info(node_list, storage_units, hvspecs)
483
485 """Compute the generic allocator input data.
486
487 @type disk_template: list of string
488 @param disk_template: the disk templates of the instances to be allocated
489
490 """
491 cfg = self.cfg.GetDetachedConfig()
492 cluster_info = cfg.GetClusterInfo()
493
494 data = {
495 "version": constants.IALLOCATOR_VERSION,
496 "cluster_name": cluster_info.cluster_name,
497 "cluster_tags": list(cluster_info.GetTags()),
498 "enabled_hypervisors": list(cluster_info.enabled_hypervisors),
499 "ipolicy": cluster_info.ipolicy,
500 }
501 ginfo = cfg.GetAllNodeGroupsInfo()
502 ninfo = cfg.GetAllNodesInfo()
503 iinfo = cfg.GetAllInstancesInfo()
504 i_list = [(inst, cluster_info.FillBE(inst)) for inst in iinfo.values()]
505
506
507 node_list = [n.uuid for n in ninfo.values() if n.vm_capable]
508
509 if isinstance(self.req, IAReqInstanceAlloc):
510 hypervisor_name = self.req.hypervisor
511 elif isinstance(self.req, IAReqRelocate):
512 hypervisor_name = iinfo[self.req.inst_uuid].hypervisor
513 else:
514 hypervisor_name = cluster_info.primary_hypervisor
515
516 if not disk_template:
517 disk_template = cluster_info.enabled_disk_templates[0]
518
519 node_data = self._ComputeClusterDataNodeInfo([disk_template], node_list,
520 cluster_info, hypervisor_name)
521
522 node_iinfo = \
523 self.rpc.call_all_instances_info(node_list,
524 cluster_info.enabled_hypervisors,
525 cluster_info.hvparams)
526
527 data["nodegroups"] = self._ComputeNodeGroupData(cluster_info, ginfo)
528
529 config_ndata = self._ComputeBasicNodeData(cfg, ninfo)
530 data["nodes"] = self._ComputeDynamicNodeData(
531 ninfo, node_data, node_iinfo, i_list, config_ndata, disk_template)
532 assert len(data["nodes"]) == len(ninfo), \
533 "Incomplete node data computed"
534
535 data["instances"] = self._ComputeInstanceData(cfg, cluster_info, i_list)
536
537 self.in_data = data
538
539 @staticmethod
541 """Compute node groups data.
542
543 """
544 ng = dict((guuid, {
545 "name": gdata.name,
546 "alloc_policy": gdata.alloc_policy,
547 "networks": [net_uuid for net_uuid, _ in gdata.networks.items()],
548 "ipolicy": gmi.CalculateGroupIPolicy(cluster, gdata),
549 "tags": list(gdata.GetTags()),
550 })
551 for guuid, gdata in ginfo.items())
552
553 return ng
554
555 @staticmethod
557 """Compute global node data.
558
559 @rtype: dict
560 @returns: a dict of name: (node dict, node config)
561
562 """
563
564 node_results = dict((ninfo.name, {
565 "tags": list(ninfo.GetTags()),
566 "primary_ip": ninfo.primary_ip,
567 "secondary_ip": ninfo.secondary_ip,
568 "offline": ninfo.offline,
569 "drained": ninfo.drained,
570 "master_candidate": ninfo.master_candidate,
571 "group": ninfo.group,
572 "master_capable": ninfo.master_capable,
573 "vm_capable": ninfo.vm_capable,
574 "ndparams": cfg.GetNdParams(ninfo),
575 })
576 for ninfo in node_cfg.values())
577
578 return node_results
579
580 @staticmethod
582 """Extract an attribute from the hypervisor's node information.
583
584 This is a helper function to extract data from the hypervisor's information
585 about the node, as part of the result of a node_info query.
586
587 @type hv_info: dict of strings
588 @param hv_info: dictionary of node information from the hypervisor
589 @type node_name: string
590 @param node_name: name of the node
591 @type attr: string
592 @param attr: key of the attribute in the hv_info dictionary
593 @rtype: integer
594 @return: the value of the attribute
595 @raises errors.OpExecError: if key not in dictionary or value not
596 integer
597
598 """
599 if attr not in hv_info:
600 raise errors.OpExecError("Node '%s' didn't return attribute"
601 " '%s'" % (node_name, attr))
602 value = hv_info[attr]
603 if not isinstance(value, int):
604 raise errors.OpExecError("Node '%s' returned invalid value"
605 " for '%s': %s" %
606 (node_name, attr, value))
607 return value
608
609 @staticmethod
612 """Extract storage data from node info.
613
614 @type space_info: see result of the RPC call node info
615 @param space_info: the storage reporting part of the result of the RPC call
616 node info
617 @type node_name: string
618 @param node_name: the node's name
619 @type disk_template: string
620 @param disk_template: the disk template to report space for
621 @rtype: 4-tuple of integers
622 @return: tuple of storage info (total_disk, free_disk, total_spindles,
623 free_spindles)
624
625 """
626 storage_type = constants.MAP_DISK_TEMPLATE_STORAGE_TYPE[disk_template]
627 if storage_type not in constants.STS_REPORT:
628 total_disk = total_spindles = 0
629 free_disk = free_spindles = 0
630 else:
631 template_space_info = utils.storage.LookupSpaceInfoByDiskTemplate(
632 space_info, disk_template)
633 if not template_space_info:
634 raise errors.OpExecError("Node '%s' didn't return space info for disk"
635 "template '%s'" % (node_name, disk_template))
636 total_disk = template_space_info["storage_size"]
637 free_disk = template_space_info["storage_free"]
638
639 total_spindles = 0
640 free_spindles = 0
641 if disk_template in constants.DTS_LVM:
642 lvm_pv_info = utils.storage.LookupSpaceInfoByStorageType(
643 space_info, constants.ST_LVM_PV)
644 if lvm_pv_info:
645 total_spindles = lvm_pv_info["storage_size"]
646 free_spindles = lvm_pv_info["storage_free"]
647 return (total_disk, free_disk, total_spindles, free_spindles)
648
649 @staticmethod
651 """Extract storage data from node info.
652
653 @type space_info: see result of the RPC call node info
654 @param space_info: the storage reporting part of the result of the RPC call
655 node info
656 @type node_name: string
657 @param node_name: the node's name
658 @type has_lvm: boolean
659 @param has_lvm: whether or not LVM storage information is requested
660 @rtype: 4-tuple of integers
661 @return: tuple of storage info (total_disk, free_disk, total_spindles,
662 free_spindles)
663
664 """
665
666 if has_lvm:
667 lvm_vg_info = utils.storage.LookupSpaceInfoByStorageType(
668 space_info, constants.ST_LVM_VG)
669 if not lvm_vg_info:
670 raise errors.OpExecError("Node '%s' didn't return LVM vg space info."
671 % (node_name))
672 total_disk = lvm_vg_info["storage_size"]
673 free_disk = lvm_vg_info["storage_free"]
674 lvm_pv_info = utils.storage.LookupSpaceInfoByStorageType(
675 space_info, constants.ST_LVM_PV)
676 if not lvm_pv_info:
677 raise errors.OpExecError("Node '%s' didn't return LVM pv space info."
678 % (node_name))
679 total_spindles = lvm_pv_info["storage_size"]
680 free_spindles = lvm_pv_info["storage_free"]
681 else:
682
683 total_disk = free_disk = 0
684 total_spindles = free_spindles = 0
685 return (total_disk, free_disk, total_spindles, free_spindles)
686
687 @staticmethod
690 """Compute memory used by primary instances.
691
692 @rtype: tuple (int, int, int)
693 @returns: A tuple of three integers: 1. the sum of memory used by primary
694 instances on the node (including the ones that are currently down), 2.
695 the sum of memory used by primary instances of the node that are up, 3.
696 the amount of memory that is free on the node considering the current
697 usage of the instances.
698
699 """
700 i_p_mem = i_p_up_mem = 0
701 mem_free = input_mem_free
702 for iinfo, beinfo in instance_list:
703 if iinfo.primary_node == node_uuid:
704 i_p_mem += beinfo[constants.BE_MAXMEM]
705 if iinfo.name not in node_instances_info[node_uuid].payload:
706 i_used_mem = 0
707 else:
708 i_used_mem = int(node_instances_info[node_uuid]
709 .payload[iinfo.name]["memory"])
710 i_mem_diff = beinfo[constants.BE_MAXMEM] - i_used_mem
711 if iinfo.admin_state == constants.ADMINST_UP \
712 and not iinfo.forthcoming:
713 mem_free -= max(0, i_mem_diff)
714 i_p_up_mem += beinfo[constants.BE_MAXMEM]
715 return (i_p_mem, i_p_up_mem, mem_free)
716
719 """Compute global node data.
720
721 @param node_results: the basic node structures as filled from the config
722
723 """
724
725
726 node_results = dict(node_results)
727 for nuuid, nresult in node_data.items():
728 ninfo = node_cfg[nuuid]
729 assert ninfo.name in node_results, "Missing basic data for node %s" % \
730 ninfo.name
731
732 if not ninfo.offline:
733 nresult.Raise("Can't get data for node %s" % ninfo.name)
734 node_iinfo[nuuid].Raise("Can't get node instance info from node %s" %
735 ninfo.name)
736 (_, space_info, (hv_info, )) = nresult.payload
737
738 mem_free = self._GetAttributeFromHypervisorNodeData(hv_info, ninfo.name,
739 "memory_free")
740
741 (i_p_mem, i_p_up_mem, mem_free) = self._ComputeInstanceMemory(
742 i_list, node_iinfo, nuuid, mem_free)
743 (total_disk, free_disk, total_spindles, free_spindles) = \
744 self._ComputeStorageDataFromSpaceInfoByTemplate(
745 space_info, ninfo.name, disk_template)
746
747
748 pnr_dyn = {
749 "total_memory": self._GetAttributeFromHypervisorNodeData(
750 hv_info, ninfo.name, "memory_total"),
751 "reserved_memory": self._GetAttributeFromHypervisorNodeData(
752 hv_info, ninfo.name, "memory_dom0"),
753 "free_memory": mem_free,
754 "total_disk": total_disk,
755 "free_disk": free_disk,
756 "total_spindles": total_spindles,
757 "free_spindles": free_spindles,
758 "total_cpus": self._GetAttributeFromHypervisorNodeData(
759 hv_info, ninfo.name, "cpu_total"),
760 "reserved_cpus": self._GetAttributeFromHypervisorNodeData(
761 hv_info, ninfo.name, "cpu_dom0"),
762 "i_pri_memory": i_p_mem,
763 "i_pri_up_memory": i_p_up_mem,
764 }
765 pnr_dyn.update(node_results[ninfo.name])
766 node_results[ninfo.name] = pnr_dyn
767
768 return node_results
769
770 @staticmethod
772 """Compute global instance data.
773
774 """
775 instance_data = {}
776 for iinfo, beinfo in i_list:
777 nic_data = []
778 for nic in iinfo.nics:
779 filled_params = cluster_info.SimpleFillNIC(nic.nicparams)
780 nic_dict = {
781 "mac": nic.mac,
782 "ip": nic.ip,
783 "mode": filled_params[constants.NIC_MODE],
784 "link": filled_params[constants.NIC_LINK],
785 }
786 if filled_params[constants.NIC_MODE] == constants.NIC_MODE_BRIDGED:
787 nic_dict["bridge"] = filled_params[constants.NIC_LINK]
788 nic_data.append(nic_dict)
789 inst_disks = cfg.GetInstanceDisks(iinfo.uuid)
790 inst_disktemplate = cfg.GetInstanceDiskTemplate(iinfo.uuid)
791 pir = {
792 "tags": list(iinfo.GetTags()),
793 "admin_state": iinfo.admin_state,
794 "vcpus": beinfo[constants.BE_VCPUS],
795 "memory": beinfo[constants.BE_MAXMEM],
796 "spindle_use": beinfo[constants.BE_SPINDLE_USE],
797 "os": iinfo.os,
798 "nodes": [cfg.GetNodeName(iinfo.primary_node)] +
799 cfg.GetNodeNames(
800 cfg.GetInstanceSecondaryNodes(iinfo.uuid)),
801 "nics": nic_data,
802 "disks": [{constants.IDISK_TYPE: dsk.dev_type,
803 constants.IDISK_SIZE: dsk.size,
804 constants.IDISK_MODE: dsk.mode,
805 constants.IDISK_SPINDLES: dsk.spindles}
806 for dsk in inst_disks],
807 "disk_template": inst_disktemplate,
808 "disks_active": iinfo.disks_active,
809 "hypervisor": iinfo.hypervisor,
810 }
811 pir["disk_space_total"] = gmi.ComputeDiskSize(pir["disks"])
812 instance_data[iinfo.name] = pir
813
814 return instance_data
815
841
842 - def Run(self, name, validate=True, call_fn=None):
843 """Run an instance allocator and return the results.
844
845 """
846 if call_fn is None:
847 call_fn = self.rpc.call_iallocator_runner
848
849 ial_params = self.cfg.GetDefaultIAllocatorParameters()
850
851 for ial_param in self.req.GetExtraParams().items():
852 ial_params[ial_param[0]] = ial_param[1]
853
854 result = call_fn(self.cfg.GetMasterNode(), name, self.in_text, ial_params)
855 result.Raise("Failure while running the iallocator script")
856
857 self.out_text = result.payload
858 if validate:
859 self._ValidateResult()
860
862 """Process the allocator results.
863
864 This will process and if successful save the result in
865 self.out_data and the other parameters.
866
867 """
868 try:
869 rdict = serializer.Load(self.out_text)
870 except Exception, err:
871 raise errors.OpExecError("Can't parse iallocator results: %s" % str(err))
872
873 if not isinstance(rdict, dict):
874 raise errors.OpExecError("Can't parse iallocator results: not a dict")
875
876
877 if "nodes" in rdict and "result" not in rdict:
878 rdict["result"] = rdict["nodes"]
879 del rdict["nodes"]
880
881 for key in "success", "info", "result":
882 if key not in rdict:
883 raise errors.OpExecError("Can't parse iallocator results:"
884 " missing key '%s'" % key)
885 setattr(self, key, rdict[key])
886
887 self.req.ValidateResult(self, self.result)
888 self.out_data = rdict
889