Fix evacuation out of drained node
[ganeti-github.git] / lib / masterd / iallocator.py
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