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
257
260 """A relocation request.
261
262 """
263
264 MODE = constants.IALLOCATOR_MODE_RELOC
265 REQ_PARAMS = [
266 _INST_UUID,
267 ("relocate_from_node_uuids", _STRING_LIST),
268 ]
269 REQ_RESULT = ht.TList
270
304
328
329 @staticmethod
331 """Returns a list of unique group names for a list of nodes.
332
333 @type node2group: dict
334 @param node2group: Map from node name to group UUID
335 @type groups: dict
336 @param groups: Group information
337 @type nodes: list
338 @param nodes: Node names
339
340 """
341 result = set()
342
343 for node in nodes:
344 try:
345 group_uuid = node2group[node]
346 except KeyError:
347
348 pass
349 else:
350 try:
351 group = groups[group_uuid]
352 except KeyError:
353
354 group_name = group_uuid
355 else:
356 group_name = group["name"]
357
358 result.add(group_name)
359
360 return sorted(result)
361
364 """A node evacuation request.
365
366 """
367
368 MODE = constants.IALLOCATOR_MODE_NODE_EVAC
369 REQ_PARAMS = [
370 ("instances", _STRING_LIST),
371 ("evac_mode", ht.TEvacMode),
372 ("ignore_soft_errors", ht.TMaybe(ht.TBool)),
373 ]
374 REQ_RESULT = _NEVAC_RESULT
375
377 """Get data for node-evacuate requests.
378
379 """
380 return {
381 "instances": self.instances,
382 "evac_mode": self.evac_mode,
383 }
384
386 """Get extra iallocator command line options for
387 node-evacuate requests.
388
389 """
390 if self.ignore_soft_errors:
391 return {"ignore-soft-errors": None}
392 else:
393 return {}
394
416
419 """IAllocator framework.
420
421 An IAllocator instance has three sets of attributes:
422 - cfg that is needed to query the cluster
423 - input data (all members of the _KEYS class attribute are required)
424 - four buffer attributes (in|out_data|text), that represent the
425 input (to the external script) in text and data structure format,
426 and the output from it, again in two formats
427 - the result variables from the script (success, info, nodes) for
428 easy usage
429
430 """
431
432
433
434 - def __init__(self, cfg, rpc_runner, req):
435 self.cfg = cfg
436 self.rpc = rpc_runner
437 self.req = req
438
439 self.in_text = self.out_text = self.in_data = self.out_data = None
440
441 self.success = self.info = self.result = None
442
443 self._BuildInputData(req)
444
447 """Prepare and execute node info call.
448
449 @type disk_templates: list of string
450 @param disk_templates: the disk templates of the instances to be allocated
451 @type node_list: list of strings
452 @param node_list: list of nodes' UUIDs
453 @type cluster_info: L{objects.Cluster}
454 @param cluster_info: the cluster's information from the config
455 @type hypervisor_name: string
456 @param hypervisor_name: the hypervisor name
457 @rtype: same as the result of the node info RPC call
458 @return: the result of the node info RPC call
459
460 """
461 storage_units_raw = utils.storage.GetStorageUnits(self.cfg, disk_templates)
462 storage_units = rpc.PrepareStorageUnitsForNodes(self.cfg, storage_units_raw,
463 node_list)
464 hvspecs = [(hypervisor_name, cluster_info.hvparams[hypervisor_name])]
465 return self.rpc.call_node_info(node_list, storage_units, hvspecs)
466
468 """Compute the generic allocator input data.
469
470 @type disk_template: list of string
471 @param disk_template: the disk templates of the instances to be allocated
472
473 """
474 cfg = self.cfg.GetDetachedConfig()
475 cluster_info = cfg.GetClusterInfo()
476
477 data = {
478 "version": constants.IALLOCATOR_VERSION,
479 "cluster_name": cluster_info.cluster_name,
480 "cluster_tags": list(cluster_info.GetTags()),
481 "enabled_hypervisors": list(cluster_info.enabled_hypervisors),
482 "ipolicy": cluster_info.ipolicy,
483 }
484 ginfo = cfg.GetAllNodeGroupsInfo()
485 ninfo = cfg.GetAllNodesInfo()
486 iinfo = cfg.GetAllInstancesInfo()
487 i_list = [(inst, cluster_info.FillBE(inst)) for inst in iinfo.values()]
488
489
490 node_list = [n.uuid for n in ninfo.values() if n.vm_capable]
491
492 if isinstance(self.req, IAReqInstanceAlloc):
493 hypervisor_name = self.req.hypervisor
494 node_whitelist = self.req.node_whitelist
495 elif isinstance(self.req, IAReqRelocate):
496 hypervisor_name = iinfo[self.req.inst_uuid].hypervisor
497 node_whitelist = None
498 else:
499 hypervisor_name = cluster_info.primary_hypervisor
500 node_whitelist = None
501
502 if not disk_template:
503 disk_template = cluster_info.enabled_disk_templates[0]
504
505 node_data = self._ComputeClusterDataNodeInfo([disk_template], node_list,
506 cluster_info, hypervisor_name)
507
508 node_iinfo = \
509 self.rpc.call_all_instances_info(node_list,
510 cluster_info.enabled_hypervisors,
511 cluster_info.hvparams)
512
513 data["nodegroups"] = self._ComputeNodeGroupData(cluster_info, ginfo)
514
515 config_ndata = self._ComputeBasicNodeData(cfg, ninfo, node_whitelist)
516 data["nodes"] = self._ComputeDynamicNodeData(
517 ninfo, node_data, node_iinfo, i_list, config_ndata, disk_template)
518 assert len(data["nodes"]) == len(ninfo), \
519 "Incomplete node data computed"
520
521 data["instances"] = self._ComputeInstanceData(cfg, cluster_info, i_list)
522
523 self.in_data = data
524
525 @staticmethod
527 """Compute node groups data.
528
529 """
530 ng = dict((guuid, {
531 "name": gdata.name,
532 "alloc_policy": gdata.alloc_policy,
533 "networks": [net_uuid for net_uuid, _ in gdata.networks.items()],
534 "ipolicy": gmi.CalculateGroupIPolicy(cluster, gdata),
535 "tags": list(gdata.GetTags()),
536 })
537 for guuid, gdata in ginfo.items())
538
539 return ng
540
541 @staticmethod
543 """Compute global node data.
544
545 @rtype: dict
546 @returns: a dict of name: (node dict, node config)
547
548 """
549
550 node_results = dict((ninfo.name, {
551 "tags": list(ninfo.GetTags()),
552 "primary_ip": ninfo.primary_ip,
553 "secondary_ip": ninfo.secondary_ip,
554 "offline": (ninfo.offline or
555 not (node_whitelist is None or
556 ninfo.name in node_whitelist)),
557 "drained": ninfo.drained,
558 "master_candidate": ninfo.master_candidate,
559 "group": ninfo.group,
560 "master_capable": ninfo.master_capable,
561 "vm_capable": ninfo.vm_capable,
562 "ndparams": cfg.GetNdParams(ninfo),
563 })
564 for ninfo in node_cfg.values())
565
566 return node_results
567
568 @staticmethod
570 """Extract an attribute from the hypervisor's node information.
571
572 This is a helper function to extract data from the hypervisor's information
573 about the node, as part of the result of a node_info query.
574
575 @type hv_info: dict of strings
576 @param hv_info: dictionary of node information from the hypervisor
577 @type node_name: string
578 @param node_name: name of the node
579 @type attr: string
580 @param attr: key of the attribute in the hv_info dictionary
581 @rtype: integer
582 @return: the value of the attribute
583 @raises errors.OpExecError: if key not in dictionary or value not
584 integer
585
586 """
587 if attr not in hv_info:
588 raise errors.OpExecError("Node '%s' didn't return attribute"
589 " '%s'" % (node_name, attr))
590 value = hv_info[attr]
591 if not isinstance(value, int):
592 raise errors.OpExecError("Node '%s' returned invalid value"
593 " for '%s': %s" %
594 (node_name, attr, value))
595 return value
596
597 @staticmethod
600 """Extract storage data from node info.
601
602 @type space_info: see result of the RPC call node info
603 @param space_info: the storage reporting part of the result of the RPC call
604 node info
605 @type node_name: string
606 @param node_name: the node's name
607 @type disk_template: string
608 @param disk_template: the disk template to report space for
609 @rtype: 4-tuple of integers
610 @return: tuple of storage info (total_disk, free_disk, total_spindles,
611 free_spindles)
612
613 """
614 storage_type = constants.MAP_DISK_TEMPLATE_STORAGE_TYPE[disk_template]
615 if storage_type not in constants.STS_REPORT:
616 total_disk = total_spindles = 0
617 free_disk = free_spindles = 0
618 else:
619 template_space_info = utils.storage.LookupSpaceInfoByDiskTemplate(
620 space_info, disk_template)
621 if not template_space_info:
622 raise errors.OpExecError("Node '%s' didn't return space info for disk"
623 "template '%s'" % (node_name, disk_template))
624 total_disk = template_space_info["storage_size"]
625 free_disk = template_space_info["storage_free"]
626
627 total_spindles = 0
628 free_spindles = 0
629 if disk_template in constants.DTS_LVM:
630 lvm_pv_info = utils.storage.LookupSpaceInfoByStorageType(
631 space_info, constants.ST_LVM_PV)
632 if lvm_pv_info:
633 total_spindles = lvm_pv_info["storage_size"]
634 free_spindles = lvm_pv_info["storage_free"]
635 return (total_disk, free_disk, total_spindles, free_spindles)
636
637 @staticmethod
639 """Extract storage data from node info.
640
641 @type space_info: see result of the RPC call node info
642 @param space_info: the storage reporting part of the result of the RPC call
643 node info
644 @type node_name: string
645 @param node_name: the node's name
646 @type has_lvm: boolean
647 @param has_lvm: whether or not LVM storage information is requested
648 @rtype: 4-tuple of integers
649 @return: tuple of storage info (total_disk, free_disk, total_spindles,
650 free_spindles)
651
652 """
653
654 if has_lvm:
655 lvm_vg_info = utils.storage.LookupSpaceInfoByStorageType(
656 space_info, constants.ST_LVM_VG)
657 if not lvm_vg_info:
658 raise errors.OpExecError("Node '%s' didn't return LVM vg space info."
659 % (node_name))
660 total_disk = lvm_vg_info["storage_size"]
661 free_disk = lvm_vg_info["storage_free"]
662 lvm_pv_info = utils.storage.LookupSpaceInfoByStorageType(
663 space_info, constants.ST_LVM_PV)
664 if not lvm_pv_info:
665 raise errors.OpExecError("Node '%s' didn't return LVM pv space info."
666 % (node_name))
667 total_spindles = lvm_pv_info["storage_size"]
668 free_spindles = lvm_pv_info["storage_free"]
669 else:
670
671 total_disk = free_disk = 0
672 total_spindles = free_spindles = 0
673 return (total_disk, free_disk, total_spindles, free_spindles)
674
675 @staticmethod
678 """Compute memory used by primary instances.
679
680 @rtype: tuple (int, int, int)
681 @returns: A tuple of three integers: 1. the sum of memory used by primary
682 instances on the node (including the ones that are currently down), 2.
683 the sum of memory used by primary instances of the node that are up, 3.
684 the amount of memory that is free on the node considering the current
685 usage of the instances.
686
687 """
688 i_p_mem = i_p_up_mem = 0
689 mem_free = input_mem_free
690 for iinfo, beinfo in instance_list:
691 if iinfo.primary_node == node_uuid:
692 i_p_mem += beinfo[constants.BE_MAXMEM]
693 if iinfo.name not in node_instances_info[node_uuid].payload:
694 i_used_mem = 0
695 else:
696 i_used_mem = int(node_instances_info[node_uuid]
697 .payload[iinfo.name]["memory"])
698 i_mem_diff = beinfo[constants.BE_MAXMEM] - i_used_mem
699 if iinfo.admin_state == constants.ADMINST_UP \
700 and not iinfo.forthcoming:
701 mem_free -= max(0, i_mem_diff)
702 i_p_up_mem += beinfo[constants.BE_MAXMEM]
703 return (i_p_mem, i_p_up_mem, mem_free)
704
707 """Compute global node data.
708
709 @param node_results: the basic node structures as filled from the config
710
711 """
712
713
714 node_results = dict(node_results)
715 for nuuid, nresult in node_data.items():
716 ninfo = node_cfg[nuuid]
717 assert ninfo.name in node_results, "Missing basic data for node %s" % \
718 ninfo.name
719
720 if not ninfo.offline:
721 nresult.Raise("Can't get data for node %s" % ninfo.name)
722 node_iinfo[nuuid].Raise("Can't get node instance info from node %s" %
723 ninfo.name)
724 (_, space_info, (hv_info, )) = nresult.payload
725
726 mem_free = self._GetAttributeFromHypervisorNodeData(hv_info, ninfo.name,
727 "memory_free")
728
729 (i_p_mem, i_p_up_mem, mem_free) = self._ComputeInstanceMemory(
730 i_list, node_iinfo, nuuid, mem_free)
731 (total_disk, free_disk, total_spindles, free_spindles) = \
732 self._ComputeStorageDataFromSpaceInfoByTemplate(
733 space_info, ninfo.name, disk_template)
734
735
736 pnr_dyn = {
737 "total_memory": self._GetAttributeFromHypervisorNodeData(
738 hv_info, ninfo.name, "memory_total"),
739 "reserved_memory": self._GetAttributeFromHypervisorNodeData(
740 hv_info, ninfo.name, "memory_dom0"),
741 "free_memory": mem_free,
742 "total_disk": total_disk,
743 "free_disk": free_disk,
744 "total_spindles": total_spindles,
745 "free_spindles": free_spindles,
746 "total_cpus": self._GetAttributeFromHypervisorNodeData(
747 hv_info, ninfo.name, "cpu_total"),
748 "reserved_cpus": self._GetAttributeFromHypervisorNodeData(
749 hv_info, ninfo.name, "cpu_dom0"),
750 "i_pri_memory": i_p_mem,
751 "i_pri_up_memory": i_p_up_mem,
752 }
753 pnr_dyn.update(node_results[ninfo.name])
754 node_results[ninfo.name] = pnr_dyn
755
756 return node_results
757
758 @staticmethod
760 """Compute global instance data.
761
762 """
763 instance_data = {}
764 for iinfo, beinfo in i_list:
765 nic_data = []
766 for nic in iinfo.nics:
767 filled_params = cluster_info.SimpleFillNIC(nic.nicparams)
768 nic_dict = {
769 "mac": nic.mac,
770 "ip": nic.ip,
771 "mode": filled_params[constants.NIC_MODE],
772 "link": filled_params[constants.NIC_LINK],
773 }
774 if filled_params[constants.NIC_MODE] == constants.NIC_MODE_BRIDGED:
775 nic_dict["bridge"] = filled_params[constants.NIC_LINK]
776 nic_data.append(nic_dict)
777 inst_disks = cfg.GetInstanceDisks(iinfo.uuid)
778 inst_disktemplate = cfg.GetInstanceDiskTemplate(iinfo.uuid)
779 pir = {
780 "tags": list(iinfo.GetTags()),
781 "admin_state": iinfo.admin_state,
782 "vcpus": beinfo[constants.BE_VCPUS],
783 "memory": beinfo[constants.BE_MAXMEM],
784 "spindle_use": beinfo[constants.BE_SPINDLE_USE],
785 "os": iinfo.os,
786 "nodes": [cfg.GetNodeName(iinfo.primary_node)] +
787 cfg.GetNodeNames(
788 cfg.GetInstanceSecondaryNodes(iinfo.uuid)),
789 "nics": nic_data,
790 "disks": [{constants.IDISK_TYPE: dsk.dev_type,
791 constants.IDISK_SIZE: dsk.size,
792 constants.IDISK_MODE: dsk.mode,
793 constants.IDISK_SPINDLES: dsk.spindles}
794 for dsk in inst_disks],
795 "disk_template": inst_disktemplate,
796 "disks_active": iinfo.disks_active,
797 "hypervisor": iinfo.hypervisor,
798 }
799 pir["disk_space_total"] = gmi.ComputeDiskSize(pir["disks"])
800 instance_data[iinfo.name] = pir
801
802 return instance_data
803
821
822 - def Run(self, name, validate=True, call_fn=None):
823 """Run an instance allocator and return the results.
824
825 """
826 if call_fn is None:
827 call_fn = self.rpc.call_iallocator_runner
828
829 ial_params = self.cfg.GetDefaultIAllocatorParameters()
830
831 for ial_param in self.req.GetExtraParams().items():
832 ial_params[ial_param[0]] = ial_param[1]
833
834 result = call_fn(self.cfg.GetMasterNode(), name, self.in_text, ial_params)
835 result.Raise("Failure while running the iallocator script")
836
837 self.out_text = result.payload
838 if validate:
839 self._ValidateResult()
840
842 """Process the allocator results.
843
844 This will process and if successful save the result in
845 self.out_data and the other parameters.
846
847 """
848 try:
849 rdict = serializer.Load(self.out_text)
850 except Exception, err:
851 raise errors.OpExecError("Can't parse iallocator results: %s" % str(err))
852
853 if not isinstance(rdict, dict):
854 raise errors.OpExecError("Can't parse iallocator results: not a dict")
855
856
857 if "nodes" in rdict and "result" not in rdict:
858 rdict["result"] = rdict["nodes"]
859 del rdict["nodes"]
860
861 for key in "success", "info", "result":
862 if key not in rdict:
863 raise errors.OpExecError("Can't parse iallocator results:"
864 " missing key '%s'" % key)
865 setattr(self, key, rdict[key])
866
867 self.req.ValidateResult(self, self.result)
868 self.out_data = rdict
869