ganeti.masterd.iallocator

1 # 2 # 3 4 # Copyright (C) 2012, 2013 Google Inc. 5 # 6 # This program is free software; you can redistribute it and/or modify 7 # it under the terms of the GNU General Public License as published by 8 # the Free Software Foundation; either version 2 of the License, or 9 # (at your option) any later version. 10 # 11 # This program is distributed in the hope that it will be useful, but 12 # WITHOUT ANY WARRANTY; without even the implied warranty of 13 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU 14 # General Public License for more details. 15 # 16 # You should have received a copy of the GNU General Public License 17 # along with this program; if not, write to the Free Software 18 # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 19 # 02110-1301, USA. 20 21 22 """Module implementing the iallocator code.""" 23 24 from ganeti import compat 25 from ganeti import constants 26 from ganeti import errors 27 from ganeti import ht 28 from ganeti import outils 29 from ganeti import opcodes 30 from ganeti import rpc 31 from ganeti import serializer 32 from ganeti import utils 33 34 import ganeti.masterd.instance as gmi 35 36 37 _STRING_LIST = ht.TListOf(ht.TString) 38 _JOB_LIST = ht.TListOf(ht.TListOf(ht.TStrictDict(True, False, { 39 # pylint: disable=E1101 40 # Class '...' has no 'OP_ID' member 41 "OP_ID": ht.TElemOf([opcodes.OpInstanceFailover.OP_ID, 42 opcodes.OpInstanceMigrate.OP_ID, 43 opcodes.OpInstanceReplaceDisks.OP_ID]), 44 }))) 45 46 _NEVAC_MOVED = \ 47 ht.TListOf(ht.TAnd(ht.TIsLength(3), 48 ht.TItems([ht.TNonEmptyString, 49 ht.TNonEmptyString, 50 ht.TListOf(ht.TNonEmptyString), 51 ]))) 52 _NEVAC_FAILED = \ 53 ht.TListOf(ht.TAnd(ht.TIsLength(2), 54 ht.TItems([ht.TNonEmptyString, 55 ht.TMaybeString, 56 ]))) 57 _NEVAC_RESULT = ht.TAnd(ht.TIsLength(3), 58 ht.TItems([_NEVAC_MOVED, _NEVAC_FAILED, _JOB_LIST])) 59 60 _INST_NAME = ("name", ht.TNonEmptyString)

61 62 63 -class _AutoReqParam(outils.AutoSlots):

64 """Meta class for request definitions. 65 66 """ 67 @classmethod

68 - def _GetSlots(mcs, attrs):

69 """Extract the slots out of REQ_PARAMS. 70 71 """ 72 params = attrs.setdefault("REQ_PARAMS", []) 73 return [slot for (slot, _) in params]

74

75 76 -class IARequestBase(outils.ValidatedSlots):

77 """A generic IAllocator request object. 78 79 """ 80 __metaclass__ = _AutoReqParam 81 82 MODE = NotImplemented 83 REQ_PARAMS = [] 84 REQ_RESULT = NotImplemented 85

86 - def __init__(self, **kwargs):

87 """Constructor for IARequestBase. 88 89 The constructor takes only keyword arguments and will set 90 attributes on this object based on the passed arguments. As such, 91 it means that you should not pass arguments which are not in the 92 REQ_PARAMS attribute for this class. 93 94 """ 95 outils.ValidatedSlots.__init__(self, **kwargs) 96 97 self.Validate()

98

99 - def Validate(self):

100 """Validates all parameters of the request. 101 102 """ 103 assert self.MODE in constants.VALID_IALLOCATOR_MODES 104 105 for (param, validator) in self.REQ_PARAMS: 106 if not hasattr(self, param): 107 raise errors.OpPrereqError("Request is missing '%s' parameter" % param, 108 errors.ECODE_INVAL) 109 110 value = getattr(self, param) 111 if not validator(value): 112 raise errors.OpPrereqError(("Request parameter '%s' has invalid" 113 " type %s/value %s") % 114 (param, type(value), value), 115 errors.ECODE_INVAL)

116

117 - def GetRequest(self, cfg):

118 """Gets the request data dict. 119 120 @param cfg: The configuration instance 121 122 """ 123 raise NotImplementedError

124

125 - def ValidateResult(self, ia, result):

126 """Validates the result of an request. 127 128 @param ia: The IAllocator instance 129 @param result: The IAllocator run result 130 @raises ResultValidationError: If validation fails 131 132 """ 133 if ia.success and not self.REQ_RESULT(result): 134 raise errors.ResultValidationError("iallocator returned invalid result," 135 " expected %s, got %s" % 136 (self.REQ_RESULT, result))

137

138 139 -class IAReqInstanceAlloc(IARequestBase):

140 """An instance allocation request. 141 142 """ 143 # pylint: disable=E1101 144 MODE = constants.IALLOCATOR_MODE_ALLOC 145 REQ_PARAMS = [ 146 _INST_NAME, 147 ("memory", ht.TNonNegativeInt), 148 ("spindle_use", ht.TNonNegativeInt), 149 ("disks", ht.TListOf(ht.TDict)), 150 ("disk_template", ht.TString), 151 ("os", ht.TString), 152 ("tags", _STRING_LIST), 153 ("nics", ht.TListOf(ht.TDict)), 154 ("vcpus", ht.TInt), 155 ("hypervisor", ht.TString), 156 ("node_whitelist", ht.TMaybeListOf(ht.TNonEmptyString)), 157 ] 158 REQ_RESULT = ht.TList 159

160 - def RequiredNodes(self):

161 """Calculates the required nodes based on the disk_template. 162 163 """ 164 if self.disk_template in constants.DTS_INT_MIRROR: 165 return 2 166 else: 167 return 1

168

169 - def GetRequest(self, cfg):

170 """Requests a new instance. 171 172 The checks for the completeness of the opcode must have already been 173 done. 174 175 """ 176 disk_space = gmi.ComputeDiskSize(self.disk_template, self.disks) 177 178 return { 179 "name": self.name, 180 "disk_template": self.disk_template, 181 "tags": self.tags, 182 "os": self.os, 183 "vcpus": self.vcpus, 184 "memory": self.memory, 185 "spindle_use": self.spindle_use, 186 "disks": self.disks, 187 "disk_space_total": disk_space, 188 "nics": self.nics, 189 "required_nodes": self.RequiredNodes(), 190 "hypervisor": self.hypervisor, 191 }

192

193 - def ValidateResult(self, ia, result):

194 """Validates an single instance allocation request. 195 196 """ 197 IARequestBase.ValidateResult(self, ia, result) 198 199 if ia.success and len(result) != self.RequiredNodes(): 200 raise errors.ResultValidationError("iallocator returned invalid number" 201 " of nodes (%s), required %s" % 202 (len(result), self.RequiredNodes()))

203

204 205 -class IAReqMultiInstanceAlloc(IARequestBase):

206 """An multi instance allocation request. 207 208 """ 209 # pylint: disable=E1101 210 MODE = constants.IALLOCATOR_MODE_MULTI_ALLOC 211 REQ_PARAMS = [ 212 ("instances", ht.TListOf(ht.TInstanceOf(IAReqInstanceAlloc))), 213 ] 214 _MASUCCESS = \ 215 ht.TListOf(ht.TAnd(ht.TIsLength(2), 216 ht.TItems([ht.TNonEmptyString, 217 ht.TListOf(ht.TNonEmptyString), 218 ]))) 219 _MAFAILED = ht.TListOf(ht.TNonEmptyString) 220 REQ_RESULT = ht.TAnd(ht.TList, ht.TIsLength(2), 221 ht.TItems([_MASUCCESS, _MAFAILED])) 222

223 - def GetRequest(self, cfg):

224 return { 225 "instances": [iareq.GetRequest(cfg) for iareq in self.instances], 226 }

227

228 229 -class IAReqRelocate(IARequestBase):

230 """A relocation request. 231 232 """ 233 # pylint: disable=E1101 234 MODE = constants.IALLOCATOR_MODE_RELOC 235 REQ_PARAMS = [ 236 _INST_NAME, 237 ("relocate_from", _STRING_LIST), 238 ] 239 REQ_RESULT = ht.TList 240

241 - def GetRequest(self, cfg):

242 """Request an relocation of an instance 243 244 The checks for the completeness of the opcode must have already been 245 done. 246 247 """ 248 instance = cfg.GetInstanceInfo(self.name) 249 if instance is None: 250 raise errors.ProgrammerError("Unknown instance '%s' passed to" 251 " IAllocator" % self.name) 252 253 if instance.disk_template not in constants.DTS_MIRRORED: 254 raise errors.OpPrereqError("Can't relocate non-mirrored instances", 255 errors.ECODE_INVAL) 256 257 if (instance.disk_template in constants.DTS_INT_MIRROR and 258 len(instance.secondary_nodes) != 1): 259 raise errors.OpPrereqError("Instance has not exactly one secondary node", 260 errors.ECODE_STATE) 261 262 disk_sizes = [{constants.IDISK_SIZE: disk.size} for disk in instance.disks] 263 disk_space = gmi.ComputeDiskSize(instance.disk_template, disk_sizes) 264 265 return { 266 "name": self.name, 267 "disk_space_total": disk_space, 268 "required_nodes": 1, 269 "relocate_from": self.relocate_from, 270 }

271

272 - def ValidateResult(self, ia, result):

273 """Validates the result of an relocation request. 274 275 """ 276 IARequestBase.ValidateResult(self, ia, result) 277 278 node2group = dict((name, ndata["group"]) 279 for (name, ndata) in ia.in_data["nodes"].items()) 280 281 fn = compat.partial(self._NodesToGroups, node2group, 282 ia.in_data["nodegroups"]) 283 284 instance = ia.cfg.GetInstanceInfo(self.name) 285 request_groups = fn(self.relocate_from + [instance.primary_node]) 286 result_groups = fn(result + [instance.primary_node]) 287 288 if ia.success and not set(result_groups).issubset(request_groups): 289 raise errors.ResultValidationError("Groups of nodes returned by" 290 "iallocator (%s) differ from original" 291 " groups (%s)" % 292 (utils.CommaJoin(result_groups), 293 utils.CommaJoin(request_groups)))

294 295 @staticmethod

296 - def _NodesToGroups(node2group, groups, nodes):

297 """Returns a list of unique group names for a list of nodes. 298 299 @type node2group: dict 300 @param node2group: Map from node name to group UUID 301 @type groups: dict 302 @param groups: Group information 303 @type nodes: list 304 @param nodes: Node names 305 306 """ 307 result = set() 308 309 for node in nodes: 310 try: 311 group_uuid = node2group[node] 312 except KeyError: 313 # Ignore unknown node 314 pass 315 else: 316 try: 317 group = groups[group_uuid] 318 except KeyError: 319 # Can't find group, let's use UUID 320 group_name = group_uuid 321 else: 322 group_name = group["name"] 323 324 result.add(group_name) 325 326 return sorted(result)

327

328 329 -class IAReqNodeEvac(IARequestBase):

330 """A node evacuation request. 331 332 """ 333 # pylint: disable=E1101 334 MODE = constants.IALLOCATOR_MODE_NODE_EVAC 335 REQ_PARAMS = [ 336 ("instances", _STRING_LIST), 337 ("evac_mode", ht.TElemOf(constants.IALLOCATOR_NEVAC_MODES)), 338 ] 339 REQ_RESULT = _NEVAC_RESULT 340

341 - def GetRequest(self, cfg):

342 """Get data for node-evacuate requests. 343 344 """ 345 return { 346 "instances": self.instances, 347 "evac_mode": self.evac_mode, 348 }

349

350 351 -class IAReqGroupChange(IARequestBase):

352 """A group change request. 353 354 """ 355 # pylint: disable=E1101 356 MODE = constants.IALLOCATOR_MODE_CHG_GROUP 357 REQ_PARAMS = [ 358 ("instances", _STRING_LIST), 359 ("target_groups", _STRING_LIST), 360 ] 361 REQ_RESULT = _NEVAC_RESULT 362

363 - def GetRequest(self, cfg):

364 """Get data for node-evacuate requests. 365 366 """ 367 return { 368 "instances": self.instances, 369 "target_groups": self.target_groups, 370 }

371

372 373 -class IAllocator(object):

374 """IAllocator framework. 375 376 An IAllocator instance has three sets of attributes: 377 - cfg that is needed to query the cluster 378 - input data (all members of the _KEYS class attribute are required) 379 - four buffer attributes (in|out_data|text), that represent the 380 input (to the external script) in text and data structure format, 381 and the output from it, again in two formats 382 - the result variables from the script (success, info, nodes) for 383 easy usage 384 385 """ 386 # pylint: disable=R0902 387 # lots of instance attributes 388

389 - def __init__(self, cfg, rpc_runner, req):

390 self.cfg = cfg 391 self.rpc = rpc_runner 392 self.req = req 393 # init buffer variables 394 self.in_text = self.out_text = self.in_data = self.out_data = None 395 # init result fields 396 self.success = self.info = self.result = None 397 398 self._BuildInputData(req)

399

400 - def _ComputeClusterData(self):

401 """Compute the generic allocator input data. 402 403 This is the data that is independent of the actual operation. 404 405 """ 406 cfg = self.cfg 407 cluster_info = cfg.GetClusterInfo() 408 # cluster data 409 data = { 410 "version": constants.IALLOCATOR_VERSION, 411 "cluster_name": cfg.GetClusterName(), 412 "cluster_tags": list(cluster_info.GetTags()), 413 "enabled_hypervisors": list(cluster_info.enabled_hypervisors), 414 "ipolicy": cluster_info.ipolicy, 415 } 416 ninfo = cfg.GetAllNodesInfo() 417 iinfo = cfg.GetAllInstancesInfo().values() 418 i_list = [(inst, cluster_info.FillBE(inst)) for inst in iinfo] 419 420 # node data 421 node_list = [n.name for n in ninfo.values() if n.vm_capable] 422 423 if isinstance(self.req, IAReqInstanceAlloc): 424 hypervisor_name = self.req.hypervisor 425 node_whitelist = self.req.node_whitelist 426 elif isinstance(self.req, IAReqRelocate): 427 hypervisor_name = cfg.GetInstanceInfo(self.req.name).hypervisor 428 node_whitelist = None 429 else: 430 hypervisor_name = cluster_info.primary_hypervisor 431 node_whitelist = None 432 433 es_flags = rpc.GetExclusiveStorageForNodeNames(cfg, node_list) 434 vg_name = cfg.GetVGName() 435 if vg_name is not None: 436 has_lvm = True 437 vg_req = [vg_name] 438 else: 439 has_lvm = False 440 vg_req = [] 441 node_data = self.rpc.call_node_info(node_list, vg_req, 442 [hypervisor_name], es_flags) 443 node_iinfo = \ 444 self.rpc.call_all_instances_info(node_list, 445 cluster_info.enabled_hypervisors) 446 447 data["nodegroups"] = self._ComputeNodeGroupData(cfg) 448 449 config_ndata = self._ComputeBasicNodeData(cfg, ninfo, node_whitelist) 450 data["nodes"] = self._ComputeDynamicNodeData(ninfo, node_data, node_iinfo, 451 i_list, config_ndata, has_lvm) 452 assert len(data["nodes"]) == len(ninfo), \ 453 "Incomplete node data computed" 454 455 data["instances"] = self._ComputeInstanceData(cluster_info, i_list) 456 457 self.in_data = data

458 459 @staticmethod

460 - def _ComputeNodeGroupData(cfg):

461 """Compute node groups data. 462 463 """ 464 cluster = cfg.GetClusterInfo() 465 ng = dict((guuid, { 466 "name": gdata.name, 467 "alloc_policy": gdata.alloc_policy, 468 "networks": [net_uuid for net_uuid, _ in gdata.networks.items()], 469 "ipolicy": gmi.CalculateGroupIPolicy(cluster, gdata), 470 "tags": list(gdata.GetTags()), 471 }) 472 for guuid, gdata in cfg.GetAllNodeGroupsInfo().items()) 473 474 return ng

475 476 @staticmethod

477 - def _ComputeBasicNodeData(cfg, node_cfg, node_whitelist):

478 """Compute global node data. 479 480 @rtype: dict 481 @returns: a dict of name: (node dict, node config) 482 483 """ 484 # fill in static (config-based) values 485 node_results = dict((ninfo.name, { 486 "tags": list(ninfo.GetTags()), 487 "primary_ip": ninfo.primary_ip, 488 "secondary_ip": ninfo.secondary_ip, 489 "offline": (ninfo.offline or 490 not (node_whitelist is None or 491 ninfo.name in node_whitelist)), 492 "drained": ninfo.drained, 493 "master_candidate": ninfo.master_candidate, 494 "group": ninfo.group, 495 "master_capable": ninfo.master_capable, 496 "vm_capable": ninfo.vm_capable, 497 "ndparams": cfg.GetNdParams(ninfo), 498 }) 499 for ninfo in node_cfg.values()) 500 501 return node_results

502 503 @staticmethod

504 - def _ComputeDynamicNodeData(node_cfg, node_data, node_iinfo, i_list, 505 node_results, has_lvm):

506 """Compute global node data. 507 508 @param node_results: the basic node structures as filled from the config 509 510 """ 511 #TODO(dynmem): compute the right data on MAX and MIN memory 512 # make a copy of the current dict 513 node_results = dict(node_results) 514 for nname, nresult in node_data.items(): 515 assert nname in node_results, "Missing basic data for node %s" % nname 516 ninfo = node_cfg[nname] 517 518 if not ninfo.offline: 519 nresult.Raise("Can't get data for node %s" % nname) 520 node_iinfo[nname].Raise("Can't get node instance info from node %s" % 521 nname) 522 remote_info = rpc.MakeLegacyNodeInfo(nresult.payload, 523 require_vg_info=has_lvm) 524 525 def get_attr(attr): 526 if attr not in remote_info: 527 raise errors.OpExecError("Node '%s' didn't return attribute" 528 " '%s'" % (nname, attr)) 529 value = remote_info[attr] 530 if not isinstance(value, int): 531 raise errors.OpExecError("Node '%s' returned invalid value" 532 " for '%s': %s" % 533 (nname, attr, value)) 534 return value

535 536 mem_free = get_attr("memory_free") 537 538 # compute memory used by primary instances 539 i_p_mem = i_p_up_mem = 0 540 for iinfo, beinfo in i_list: 541 if iinfo.primary_node == nname: 542 i_p_mem += beinfo[constants.BE_MAXMEM] 543 if iinfo.name not in node_iinfo[nname].payload: 544 i_used_mem = 0 545 else: 546 i_used_mem = int(node_iinfo[nname].payload[iinfo.name]["memory"]) 547 i_mem_diff = beinfo[constants.BE_MAXMEM] - i_used_mem 548 mem_free -= max(0, i_mem_diff) 549 550 if iinfo.admin_state == constants.ADMINST_UP: 551 i_p_up_mem += beinfo[constants.BE_MAXMEM] 552 553 # TODO: replace this with proper storage reporting 554 if has_lvm: 555 total_disk = get_attr("vg_size") 556 free_disk = get_attr("vg_free") 557 else: 558 # we didn't even ask the node for VG status, so use zeros 559 total_disk = free_disk = 0 560 561 # compute memory used by instances 562 pnr_dyn = { 563 "total_memory": get_attr("memory_total"), 564 "reserved_memory": get_attr("memory_dom0"), 565 "free_memory": mem_free, 566 "total_disk": total_disk, 567 "free_disk": free_disk, 568 "total_cpus": get_attr("cpu_total"), 569 "i_pri_memory": i_p_mem, 570 "i_pri_up_memory": i_p_up_mem, 571 } 572 pnr_dyn.update(node_results[nname]) 573 node_results[nname] = pnr_dyn 574 575 return node_results

576 577 @staticmethod

578 - def _ComputeInstanceData(cluster_info, i_list):

579 """Compute global instance data. 580 581 """ 582 instance_data = {} 583 for iinfo, beinfo in i_list: 584 nic_data = [] 585 for nic in iinfo.nics: 586 filled_params = cluster_info.SimpleFillNIC(nic.nicparams) 587 nic_dict = { 588 "mac": nic.mac, 589 "ip": nic.ip, 590 "mode": filled_params[constants.NIC_MODE], 591 "link": filled_params[constants.NIC_LINK], 592 } 593 if filled_params[constants.NIC_MODE] == constants.NIC_MODE_BRIDGED: 594 nic_dict["bridge"] = filled_params[constants.NIC_LINK] 595 nic_data.append(nic_dict) 596 pir = { 597 "tags": list(iinfo.GetTags()), 598 "admin_state": iinfo.admin_state, 599 "vcpus": beinfo[constants.BE_VCPUS], 600 "memory": beinfo[constants.BE_MAXMEM], 601 "spindle_use": beinfo[constants.BE_SPINDLE_USE], 602 "os": iinfo.os, 603 "nodes": [iinfo.primary_node] + list(iinfo.secondary_nodes), 604 "nics": nic_data, 605 "disks": [{constants.IDISK_SIZE: dsk.size, 606 constants.IDISK_MODE: dsk.mode} 607 for dsk in iinfo.disks], 608 "disk_template": iinfo.disk_template, 609 "disks_active": iinfo.disks_active, 610 "hypervisor": iinfo.hypervisor, 611 } 612 pir["disk_space_total"] = gmi.ComputeDiskSize(iinfo.disk_template, 613 pir["disks"]) 614 instance_data[iinfo.name] = pir 615 616 return instance_data

617

618 - def _BuildInputData(self, req):

619 """Build input data structures. 620 621 """ 622 self._ComputeClusterData() 623 624 request = req.GetRequest(self.cfg) 625 request["type"] = req.MODE 626 self.in_data["request"] = request 627 628 self.in_text = serializer.Dump(self.in_data)

629

630 - def Run(self, name, validate=True, call_fn=None):

631 """Run an instance allocator and return the results. 632 633 """ 634 if call_fn is None: 635 call_fn = self.rpc.call_iallocator_runner 636 637 result = call_fn(self.cfg.GetMasterNode(), name, self.in_text) 638 result.Raise("Failure while running the iallocator script") 639 640 self.out_text = result.payload 641 if validate: 642 self._ValidateResult()

643

644 - def _ValidateResult(self):

645 """Process the allocator results. 646 647 This will process and if successful save the result in 648 self.out_data and the other parameters. 649 650 """ 651 try: 652 rdict = serializer.Load(self.out_text) 653 except Exception, err: 654 raise errors.OpExecError("Can't parse iallocator results: %s" % str(err)) 655 656 if not isinstance(rdict, dict): 657 raise errors.OpExecError("Can't parse iallocator results: not a dict") 658 659 # TODO: remove backwards compatiblity in later versions 660 if "nodes" in rdict and "result" not in rdict: 661 rdict["result"] = rdict["nodes"] 662 del rdict["nodes"] 663 664 for key in "success", "info", "result": 665 if key not in rdict: 666 raise errors.OpExecError("Can't parse iallocator results:" 667 " missing key '%s'" % key) 668 setattr(self, key, rdict[key]) 669 670 self.req.ValidateResult(self, self.result) 671 self.out_data = rdict

672

Source Code for Module ganeti.masterd.iallocator