4 # Copyright (C) 2006, 2007 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 """Configuration management for Ganeti
24 This module provides the interface to the Ganeti cluster configuration.
26 The configuration data is stored on every node but is updated on the master
27 only. After each update, the master distributes the data to the other nodes.
29 Currently, the data storage format is JSON. YAML was slow and consuming too
39 from ganeti import errors
40 from ganeti import locking
41 from ganeti import utils
42 from ganeti import constants
43 from ganeti import rpc
44 from ganeti import objects
45 from ganeti import serializer
48 _config_lock = locking.SharedLock()
50 # job id used for resource management at config upgrade time
51 _UPGRADE_CONFIG_JID = "jid-cfg-upgrade"
54 def _ValidateConfig(data):
55 """Verifies that a configuration objects looks valid.
57 This only verifies the version of the configuration.
59 @raise errors.ConfigurationError: if the version differs from what
63 if data.version != constants.CONFIG_VERSION:
64 raise errors.ConfigurationError("Cluster configuration version"
65 " mismatch, got %s instead of %s" %
67 constants.CONFIG_VERSION))
70 class TemporaryReservationManager:
71 """A temporary resource reservation manager.
73 This is used to reserve resources in a job, before using them, making sure
74 other jobs cannot get them in the meantime.
78 self._ec_reserved = {}
80 def Reserved(self, resource):
81 for holder_reserved in self._ec_reserved.items():
82 if resource in holder_reserved:
86 def Reserve(self, ec_id, resource):
87 if self.Reserved(resource):
88 raise errors.ReservationError("Duplicate reservation for resource: %s." %
90 if ec_id not in self._ec_reserved:
91 self._ec_reserved[ec_id] = set([resource])
93 self._ec_reserved[ec_id].add(resource)
95 def DropECReservations(self, ec_id):
96 if ec_id in self._ec_reserved:
97 del self._ec_reserved[ec_id]
99 def GetReserved(self):
101 for holder_reserved in self._ec_reserved.values():
102 all_reserved.update(holder_reserved)
105 def Generate(self, existing, generate_one_fn, ec_id):
106 """Generate a new resource of this type
109 assert callable(generate_one_fn)
111 all_elems = self.GetReserved()
112 all_elems.update(existing)
115 new_resource = generate_one_fn()
116 if new_resource is not None and new_resource not in all_elems:
119 raise errors.ConfigurationError("Not able generate new resource"
120 " (last tried: %s)" % new_resource)
121 self.Reserve(ec_id, new_resource)
126 """The interface to the cluster configuration.
129 def __init__(self, cfg_file=None, offline=False):
131 self._lock = _config_lock
132 self._config_data = None
133 self._offline = offline
135 self._cfg_file = constants.CLUSTER_CONF_FILE
137 self._cfg_file = cfg_file
138 self._temporary_ids = TemporaryReservationManager()
139 self._temporary_drbds = {}
140 self._temporary_macs = TemporaryReservationManager()
141 self._temporary_secrets = TemporaryReservationManager()
142 # Note: in order to prevent errors when resolving our name in
143 # _DistributeConfig, we compute it here once and reuse it; it's
144 # better to raise an error before starting to modify the config
145 # file than after it was modified
146 self._my_hostname = utils.HostInfo().name
147 self._last_cluster_serial = -1
150 # this method needs to be static, so that we can call it on the class
153 """Check if the cluster is configured.
156 return os.path.exists(constants.CLUSTER_CONF_FILE)
158 def _GenerateOneMAC(self):
159 """Generate one mac address
162 prefix = self._config_data.cluster.mac_prefix
163 byte1 = random.randrange(0, 256)
164 byte2 = random.randrange(0, 256)
165 byte3 = random.randrange(0, 256)
166 mac = "%s:%02x:%02x:%02x" % (prefix, byte1, byte2, byte3)
169 @locking.ssynchronized(_config_lock, shared=1)
170 def GenerateMAC(self, ec_id):
171 """Generate a MAC for an instance.
173 This should check the current instances for duplicates.
176 existing = self._AllMACs()
177 return self._temporary_ids.Generate(existing, self._GenerateOneMAC, ec_id)
179 @locking.ssynchronized(_config_lock, shared=1)
180 def ReserveMAC(self, mac, ec_id):
181 """Reserve a MAC for an instance.
183 This only checks instances managed by this cluster, it does not
184 check for potential collisions elsewhere.
187 all_macs = self._AllMACs()
189 raise errors.ReservationError("mac already in use")
191 self._temporary_macs.Reserve(mac, ec_id)
193 @locking.ssynchronized(_config_lock, shared=1)
194 def GenerateDRBDSecret(self, ec_id):
195 """Generate a DRBD secret.
197 This checks the current disks for duplicates.
200 return self._temporary_secrets.Generate(self._AllDRBDSecrets(),
201 utils.GenerateSecret,
205 """Compute the list of all LVs.
209 for instance in self._config_data.instances.values():
210 node_data = instance.MapLVsByNode()
211 for lv_list in node_data.values():
212 lvnames.update(lv_list)
215 def _AllIDs(self, include_temporary):
216 """Compute the list of all UUIDs and names we have.
218 @type include_temporary: boolean
219 @param include_temporary: whether to include the _temporary_ids set
221 @return: a set of IDs
225 if include_temporary:
226 existing.update(self._temporary_ids.GetReserved())
227 existing.update(self._AllLVs())
228 existing.update(self._config_data.instances.keys())
229 existing.update(self._config_data.nodes.keys())
230 existing.update([i.uuid for i in self._AllUUIDObjects() if i.uuid])
233 def _GenerateUniqueID(self, ec_id):
234 """Generate an unique UUID.
236 This checks the current node, instances and disk names for
240 @return: the unique id
243 existing = self._AllIDs(include_temporary=False)
244 return self._temporary_ids.Generate(existing, utils.NewUUID, ec_id)
246 @locking.ssynchronized(_config_lock, shared=1)
247 def GenerateUniqueID(self, ec_id):
248 """Generate an unique ID.
250 This is just a wrapper over the unlocked version.
253 @param ec_id: unique id for the job to reserve the id to
256 return self._GenerateUniqueID(ec_id)
259 """Return all MACs present in the config.
262 @return: the list of all MACs
266 for instance in self._config_data.instances.values():
267 for nic in instance.nics:
268 result.append(nic.mac)
272 def _AllDRBDSecrets(self):
273 """Return all DRBD secrets present in the config.
276 @return: the list of all DRBD secrets
279 def helper(disk, result):
280 """Recursively gather secrets from this disk."""
281 if disk.dev_type == constants.DT_DRBD8:
282 result.append(disk.logical_id[5])
284 for child in disk.children:
285 helper(child, result)
288 for instance in self._config_data.instances.values():
289 for disk in instance.disks:
294 def _CheckDiskIDs(self, disk, l_ids, p_ids):
295 """Compute duplicate disk IDs
297 @type disk: L{objects.Disk}
298 @param disk: the disk at which to start searching
300 @param l_ids: list of current logical ids
302 @param p_ids: list of current physical ids
304 @return: a list of error messages
308 if disk.logical_id is not None:
309 if disk.logical_id in l_ids:
310 result.append("duplicate logical id %s" % str(disk.logical_id))
312 l_ids.append(disk.logical_id)
313 if disk.physical_id is not None:
314 if disk.physical_id in p_ids:
315 result.append("duplicate physical id %s" % str(disk.physical_id))
317 p_ids.append(disk.physical_id)
320 for child in disk.children:
321 result.extend(self._CheckDiskIDs(child, l_ids, p_ids))
324 def _UnlockedVerifyConfig(self):
328 @return: a list of error messages; a non-empty list signifies
335 data = self._config_data
339 # global cluster checks
340 if not data.cluster.enabled_hypervisors:
341 result.append("enabled hypervisors list doesn't have any entries")
342 invalid_hvs = set(data.cluster.enabled_hypervisors) - constants.HYPER_TYPES
344 result.append("enabled hypervisors contains invalid entries: %s" %
347 if data.cluster.master_node not in data.nodes:
348 result.append("cluster has invalid primary node '%s'" %
349 data.cluster.master_node)
351 # per-instance checks
352 for instance_name in data.instances:
353 instance = data.instances[instance_name]
354 if instance.primary_node not in data.nodes:
355 result.append("instance '%s' has invalid primary node '%s'" %
356 (instance_name, instance.primary_node))
357 for snode in instance.secondary_nodes:
358 if snode not in data.nodes:
359 result.append("instance '%s' has invalid secondary node '%s'" %
360 (instance_name, snode))
361 for idx, nic in enumerate(instance.nics):
362 if nic.mac in seen_macs:
363 result.append("instance '%s' has NIC %d mac %s duplicate" %
364 (instance_name, idx, nic.mac))
366 seen_macs.append(nic.mac)
368 # gather the drbd ports for duplicate checks
369 for dsk in instance.disks:
370 if dsk.dev_type in constants.LDS_DRBD:
371 tcp_port = dsk.logical_id[2]
372 if tcp_port not in ports:
374 ports[tcp_port].append((instance.name, "drbd disk %s" % dsk.iv_name))
375 # gather network port reservation
376 net_port = getattr(instance, "network_port", None)
377 if net_port is not None:
378 if net_port not in ports:
380 ports[net_port].append((instance.name, "network port"))
382 # instance disk verify
383 for idx, disk in enumerate(instance.disks):
384 result.extend(["instance '%s' disk %d error: %s" %
385 (instance.name, idx, msg) for msg in disk.Verify()])
386 result.extend(self._CheckDiskIDs(disk, seen_lids, seen_pids))
388 # cluster-wide pool of free ports
389 for free_port in data.cluster.tcpudp_port_pool:
390 if free_port not in ports:
391 ports[free_port] = []
392 ports[free_port].append(("cluster", "port marked as free"))
394 # compute tcp/udp duplicate ports
400 txt = utils.CommaJoin(["%s/%s" % val for val in pdata])
401 result.append("tcp/udp port %s has duplicates: %s" % (pnum, txt))
403 # highest used tcp port check
405 if keys[-1] > data.cluster.highest_used_port:
406 result.append("Highest used port mismatch, saved %s, computed %s" %
407 (data.cluster.highest_used_port, keys[-1]))
409 if not data.nodes[data.cluster.master_node].master_candidate:
410 result.append("Master node is not a master candidate")
412 # master candidate checks
413 mc_now, mc_max, _ = self._UnlockedGetMasterCandidateStats()
415 result.append("Not enough master candidates: actual %d, target %d" %
419 for node in data.nodes.values():
420 if [node.master_candidate, node.drained, node.offline].count(True) > 1:
421 result.append("Node %s state is invalid: master_candidate=%s,"
422 " drain=%s, offline=%s" %
423 (node.name, node.master_candidate, node.drain,
427 _, duplicates = self._UnlockedComputeDRBDMap()
428 for node, minor, instance_a, instance_b in duplicates:
429 result.append("DRBD minor %d on node %s is assigned twice to instances"
430 " %s and %s" % (minor, node, instance_a, instance_b))
433 default_nicparams = data.cluster.nicparams[constants.PP_DEFAULT]
436 def _AddIpAddress(ip, name):
437 ips.setdefault(ip, []).append(name)
439 _AddIpAddress(data.cluster.master_ip, "cluster_ip")
441 for node in data.nodes.values():
442 _AddIpAddress(node.primary_ip, "node:%s/primary" % node.name)
443 if node.secondary_ip != node.primary_ip:
444 _AddIpAddress(node.secondary_ip, "node:%s/secondary" % node.name)
446 for instance in data.instances.values():
447 for idx, nic in enumerate(instance.nics):
451 nicparams = objects.FillDict(default_nicparams, nic.nicparams)
452 nic_mode = nicparams[constants.NIC_MODE]
453 nic_link = nicparams[constants.NIC_LINK]
455 if nic_mode == constants.NIC_MODE_BRIDGED:
456 link = "bridge:%s" % nic_link
457 elif nic_mode == constants.NIC_MODE_ROUTED:
458 link = "route:%s" % nic_link
460 raise errors.ProgrammerError("NIC mode '%s' not handled" % nic_mode)
462 _AddIpAddress("%s/%s" % (link, nic.ip),
463 "instance:%s/nic:%d" % (instance.name, idx))
465 for ip, owners in ips.items():
467 result.append("IP address %s is used by multiple owners: %s" %
468 (ip, utils.CommaJoin(owners)))
472 @locking.ssynchronized(_config_lock, shared=1)
473 def VerifyConfig(self):
476 This is just a wrapper over L{_UnlockedVerifyConfig}.
479 @return: a list of error messages; a non-empty list signifies
483 return self._UnlockedVerifyConfig()
485 def _UnlockedSetDiskID(self, disk, node_name):
486 """Convert the unique ID to the ID needed on the target nodes.
488 This is used only for drbd, which needs ip/port configuration.
490 The routine descends down and updates its children also, because
491 this helps when the only the top device is passed to the remote
494 This function is for internal use, when the config lock is already held.
498 for child in disk.children:
499 self._UnlockedSetDiskID(child, node_name)
501 if disk.logical_id is None and disk.physical_id is not None:
503 if disk.dev_type == constants.LD_DRBD8:
504 pnode, snode, port, pminor, sminor, secret = disk.logical_id
505 if node_name not in (pnode, snode):
506 raise errors.ConfigurationError("DRBD device not knowing node %s" %
508 pnode_info = self._UnlockedGetNodeInfo(pnode)
509 snode_info = self._UnlockedGetNodeInfo(snode)
510 if pnode_info is None or snode_info is None:
511 raise errors.ConfigurationError("Can't find primary or secondary node"
512 " for %s" % str(disk))
513 p_data = (pnode_info.secondary_ip, port)
514 s_data = (snode_info.secondary_ip, port)
515 if pnode == node_name:
516 disk.physical_id = p_data + s_data + (pminor, secret)
517 else: # it must be secondary, we tested above
518 disk.physical_id = s_data + p_data + (sminor, secret)
520 disk.physical_id = disk.logical_id
523 @locking.ssynchronized(_config_lock)
524 def SetDiskID(self, disk, node_name):
525 """Convert the unique ID to the ID needed on the target nodes.
527 This is used only for drbd, which needs ip/port configuration.
529 The routine descends down and updates its children also, because
530 this helps when the only the top device is passed to the remote
534 return self._UnlockedSetDiskID(disk, node_name)
536 @locking.ssynchronized(_config_lock)
537 def AddTcpUdpPort(self, port):
538 """Adds a new port to the available port pool.
541 if not isinstance(port, int):
542 raise errors.ProgrammerError("Invalid type passed for port")
544 self._config_data.cluster.tcpudp_port_pool.add(port)
547 @locking.ssynchronized(_config_lock, shared=1)
548 def GetPortList(self):
549 """Returns a copy of the current port list.
552 return self._config_data.cluster.tcpudp_port_pool.copy()
554 @locking.ssynchronized(_config_lock)
555 def AllocatePort(self):
558 The port will be taken from the available port pool or from the
559 default port range (and in this case we increase
563 # If there are TCP/IP ports configured, we use them first.
564 if self._config_data.cluster.tcpudp_port_pool:
565 port = self._config_data.cluster.tcpudp_port_pool.pop()
567 port = self._config_data.cluster.highest_used_port + 1
568 if port >= constants.LAST_DRBD_PORT:
569 raise errors.ConfigurationError("The highest used port is greater"
570 " than %s. Aborting." %
571 constants.LAST_DRBD_PORT)
572 self._config_data.cluster.highest_used_port = port
577 def _UnlockedComputeDRBDMap(self):
578 """Compute the used DRBD minor/nodes.
581 @return: dictionary of node_name: dict of minor: instance_name;
582 the returned dict will have all the nodes in it (even if with
583 an empty list), and a list of duplicates; if the duplicates
584 list is not empty, the configuration is corrupted and its caller
585 should raise an exception
588 def _AppendUsedPorts(instance_name, disk, used):
590 if disk.dev_type == constants.LD_DRBD8 and len(disk.logical_id) >= 5:
591 node_a, node_b, _, minor_a, minor_b = disk.logical_id[:5]
592 for node, port in ((node_a, minor_a), (node_b, minor_b)):
593 assert node in used, ("Node '%s' of instance '%s' not found"
594 " in node list" % (node, instance_name))
595 if port in used[node]:
596 duplicates.append((node, port, instance_name, used[node][port]))
598 used[node][port] = instance_name
600 for child in disk.children:
601 duplicates.extend(_AppendUsedPorts(instance_name, child, used))
605 my_dict = dict((node, {}) for node in self._config_data.nodes)
606 for instance in self._config_data.instances.itervalues():
607 for disk in instance.disks:
608 duplicates.extend(_AppendUsedPorts(instance.name, disk, my_dict))
609 for (node, minor), instance in self._temporary_drbds.iteritems():
610 if minor in my_dict[node] and my_dict[node][minor] != instance:
611 duplicates.append((node, minor, instance, my_dict[node][minor]))
613 my_dict[node][minor] = instance
614 return my_dict, duplicates
616 @locking.ssynchronized(_config_lock)
617 def ComputeDRBDMap(self):
618 """Compute the used DRBD minor/nodes.
620 This is just a wrapper over L{_UnlockedComputeDRBDMap}.
622 @return: dictionary of node_name: dict of minor: instance_name;
623 the returned dict will have all the nodes in it (even if with
627 d_map, duplicates = self._UnlockedComputeDRBDMap()
629 raise errors.ConfigurationError("Duplicate DRBD ports detected: %s" %
633 @locking.ssynchronized(_config_lock)
634 def AllocateDRBDMinor(self, nodes, instance):
635 """Allocate a drbd minor.
637 The free minor will be automatically computed from the existing
638 devices. A node can be given multiple times in order to allocate
639 multiple minors. The result is the list of minors, in the same
640 order as the passed nodes.
642 @type instance: string
643 @param instance: the instance for which we allocate minors
646 assert isinstance(instance, basestring), \
647 "Invalid argument '%s' passed to AllocateDRBDMinor" % instance
649 d_map, duplicates = self._UnlockedComputeDRBDMap()
651 raise errors.ConfigurationError("Duplicate DRBD ports detected: %s" %
657 # no minors used, we can start at 0
660 self._temporary_drbds[(nname, 0)] = instance
664 ffree = utils.FirstFree(keys)
666 # return the next minor
667 # TODO: implement high-limit check
671 # double-check minor against current instances
672 assert minor not in d_map[nname], \
673 ("Attempt to reuse allocated DRBD minor %d on node %s,"
674 " already allocated to instance %s" %
675 (minor, nname, d_map[nname][minor]))
676 ndata[minor] = instance
677 # double-check minor against reservation
678 r_key = (nname, minor)
679 assert r_key not in self._temporary_drbds, \
680 ("Attempt to reuse reserved DRBD minor %d on node %s,"
681 " reserved for instance %s" %
682 (minor, nname, self._temporary_drbds[r_key]))
683 self._temporary_drbds[r_key] = instance
685 logging.debug("Request to allocate drbd minors, input: %s, returning %s",
689 def _UnlockedReleaseDRBDMinors(self, instance):
690 """Release temporary drbd minors allocated for a given instance.
692 @type instance: string
693 @param instance: the instance for which temporary minors should be
697 assert isinstance(instance, basestring), \
698 "Invalid argument passed to ReleaseDRBDMinors"
699 for key, name in self._temporary_drbds.items():
701 del self._temporary_drbds[key]
703 @locking.ssynchronized(_config_lock)
704 def ReleaseDRBDMinors(self, instance):
705 """Release temporary drbd minors allocated for a given instance.
707 This should be called on the error paths, on the success paths
708 it's automatically called by the ConfigWriter add and update
711 This function is just a wrapper over L{_UnlockedReleaseDRBDMinors}.
713 @type instance: string
714 @param instance: the instance for which temporary minors should be
718 self._UnlockedReleaseDRBDMinors(instance)
720 @locking.ssynchronized(_config_lock, shared=1)
721 def GetConfigVersion(self):
722 """Get the configuration version.
724 @return: Config version
727 return self._config_data.version
729 @locking.ssynchronized(_config_lock, shared=1)
730 def GetClusterName(self):
733 @return: Cluster name
736 return self._config_data.cluster.cluster_name
738 @locking.ssynchronized(_config_lock, shared=1)
739 def GetMasterNode(self):
740 """Get the hostname of the master node for this cluster.
742 @return: Master hostname
745 return self._config_data.cluster.master_node
747 @locking.ssynchronized(_config_lock, shared=1)
748 def GetMasterIP(self):
749 """Get the IP of the master node for this cluster.
754 return self._config_data.cluster.master_ip
756 @locking.ssynchronized(_config_lock, shared=1)
757 def GetMasterNetdev(self):
758 """Get the master network device for this cluster.
761 return self._config_data.cluster.master_netdev
763 @locking.ssynchronized(_config_lock, shared=1)
764 def GetFileStorageDir(self):
765 """Get the file storage dir for this cluster.
768 return self._config_data.cluster.file_storage_dir
770 @locking.ssynchronized(_config_lock, shared=1)
771 def GetHypervisorType(self):
772 """Get the hypervisor type for this cluster.
775 return self._config_data.cluster.enabled_hypervisors[0]
777 @locking.ssynchronized(_config_lock, shared=1)
778 def GetHostKey(self):
779 """Return the rsa hostkey from the config.
782 @return: the rsa hostkey
785 return self._config_data.cluster.rsahostkeypub
787 @locking.ssynchronized(_config_lock)
788 def AddInstance(self, instance, ec_id):
789 """Add an instance to the config.
791 This should be used after creating a new instance.
793 @type instance: L{objects.Instance}
794 @param instance: the instance object
797 if not isinstance(instance, objects.Instance):
798 raise errors.ProgrammerError("Invalid type passed to AddInstance")
800 if instance.disk_template != constants.DT_DISKLESS:
801 all_lvs = instance.MapLVsByNode()
802 logging.info("Instance '%s' DISK_LAYOUT: %s", instance.name, all_lvs)
804 all_macs = self._AllMACs()
805 for nic in instance.nics:
806 if nic.mac in all_macs:
807 raise errors.ConfigurationError("Cannot add instance %s:"
808 " MAC address '%s' already in use." %
809 (instance.name, nic.mac))
811 self._EnsureUUID(instance, ec_id)
813 instance.serial_no = 1
814 instance.ctime = instance.mtime = time.time()
815 self._config_data.instances[instance.name] = instance
816 self._config_data.cluster.serial_no += 1
817 self._UnlockedReleaseDRBDMinors(instance.name)
820 def _EnsureUUID(self, item, ec_id):
821 """Ensures a given object has a valid UUID.
823 @param item: the instance or node to be checked
824 @param ec_id: the execution context id for the uuid reservation
828 item.uuid = self._GenerateUniqueID(ec_id)
829 elif item.uuid in self._AllIDs(include_temporary=True):
830 raise errors.ConfigurationError("Cannot add '%s': UUID %s already"
831 " in use" % (item.name, item.uuid))
833 def _SetInstanceStatus(self, instance_name, status):
834 """Set the instance's status to a given value.
837 assert isinstance(status, bool), \
838 "Invalid status '%s' passed to SetInstanceStatus" % (status,)
840 if instance_name not in self._config_data.instances:
841 raise errors.ConfigurationError("Unknown instance '%s'" %
843 instance = self._config_data.instances[instance_name]
844 if instance.admin_up != status:
845 instance.admin_up = status
846 instance.serial_no += 1
847 instance.mtime = time.time()
850 @locking.ssynchronized(_config_lock)
851 def MarkInstanceUp(self, instance_name):
852 """Mark the instance status to up in the config.
855 self._SetInstanceStatus(instance_name, True)
857 @locking.ssynchronized(_config_lock)
858 def RemoveInstance(self, instance_name):
859 """Remove the instance from the configuration.
862 if instance_name not in self._config_data.instances:
863 raise errors.ConfigurationError("Unknown instance '%s'" % instance_name)
864 del self._config_data.instances[instance_name]
865 self._config_data.cluster.serial_no += 1
868 @locking.ssynchronized(_config_lock)
869 def RenameInstance(self, old_name, new_name):
870 """Rename an instance.
872 This needs to be done in ConfigWriter and not by RemoveInstance
873 combined with AddInstance as only we can guarantee an atomic
877 if old_name not in self._config_data.instances:
878 raise errors.ConfigurationError("Unknown instance '%s'" % old_name)
879 inst = self._config_data.instances[old_name]
880 del self._config_data.instances[old_name]
883 for disk in inst.disks:
884 if disk.dev_type == constants.LD_FILE:
885 # rename the file paths in logical and physical id
886 file_storage_dir = os.path.dirname(os.path.dirname(disk.logical_id[1]))
887 disk.physical_id = disk.logical_id = (disk.logical_id[0],
888 utils.PathJoin(file_storage_dir,
892 self._config_data.instances[inst.name] = inst
895 @locking.ssynchronized(_config_lock)
896 def MarkInstanceDown(self, instance_name):
897 """Mark the status of an instance to down in the configuration.
900 self._SetInstanceStatus(instance_name, False)
902 def _UnlockedGetInstanceList(self):
903 """Get the list of instances.
905 This function is for internal use, when the config lock is already held.
908 return self._config_data.instances.keys()
910 @locking.ssynchronized(_config_lock, shared=1)
911 def GetInstanceList(self):
912 """Get the list of instances.
914 @return: array of instances, ex. ['instance2.example.com',
915 'instance1.example.com']
918 return self._UnlockedGetInstanceList()
920 @locking.ssynchronized(_config_lock, shared=1)
921 def ExpandInstanceName(self, short_name):
922 """Attempt to expand an incomplete instance name.
925 return utils.MatchNameComponent(short_name,
926 self._config_data.instances.keys(),
927 case_sensitive=False)
929 def _UnlockedGetInstanceInfo(self, instance_name):
930 """Returns information about an instance.
932 This function is for internal use, when the config lock is already held.
935 if instance_name not in self._config_data.instances:
938 return self._config_data.instances[instance_name]
940 @locking.ssynchronized(_config_lock, shared=1)
941 def GetInstanceInfo(self, instance_name):
942 """Returns information about an instance.
944 It takes the information from the configuration file. Other information of
945 an instance are taken from the live systems.
947 @param instance_name: name of the instance, e.g.
948 I{instance1.example.com}
950 @rtype: L{objects.Instance}
951 @return: the instance object
954 return self._UnlockedGetInstanceInfo(instance_name)
956 @locking.ssynchronized(_config_lock, shared=1)
957 def GetAllInstancesInfo(self):
958 """Get the configuration of all instances.
961 @return: dict of (instance, instance_info), where instance_info is what
962 would GetInstanceInfo return for the node
965 my_dict = dict([(instance, self._UnlockedGetInstanceInfo(instance))
966 for instance in self._UnlockedGetInstanceList()])
969 @locking.ssynchronized(_config_lock)
970 def AddNode(self, node, ec_id):
971 """Add a node to the configuration.
973 @type node: L{objects.Node}
974 @param node: a Node instance
977 logging.info("Adding node %s to configuration", node.name)
979 self._EnsureUUID(node, ec_id)
982 node.ctime = node.mtime = time.time()
983 self._config_data.nodes[node.name] = node
984 self._config_data.cluster.serial_no += 1
987 @locking.ssynchronized(_config_lock)
988 def RemoveNode(self, node_name):
989 """Remove a node from the configuration.
992 logging.info("Removing node %s from configuration", node_name)
994 if node_name not in self._config_data.nodes:
995 raise errors.ConfigurationError("Unknown node '%s'" % node_name)
997 del self._config_data.nodes[node_name]
998 self._config_data.cluster.serial_no += 1
1001 @locking.ssynchronized(_config_lock, shared=1)
1002 def ExpandNodeName(self, short_name):
1003 """Attempt to expand an incomplete instance name.
1006 return utils.MatchNameComponent(short_name,
1007 self._config_data.nodes.keys(),
1008 case_sensitive=False)
1010 def _UnlockedGetNodeInfo(self, node_name):
1011 """Get the configuration of a node, as stored in the config.
1013 This function is for internal use, when the config lock is already
1016 @param node_name: the node name, e.g. I{node1.example.com}
1018 @rtype: L{objects.Node}
1019 @return: the node object
1022 if node_name not in self._config_data.nodes:
1025 return self._config_data.nodes[node_name]
1027 @locking.ssynchronized(_config_lock, shared=1)
1028 def GetNodeInfo(self, node_name):
1029 """Get the configuration of a node, as stored in the config.
1031 This is just a locked wrapper over L{_UnlockedGetNodeInfo}.
1033 @param node_name: the node name, e.g. I{node1.example.com}
1035 @rtype: L{objects.Node}
1036 @return: the node object
1039 return self._UnlockedGetNodeInfo(node_name)
1041 def _UnlockedGetNodeList(self):
1042 """Return the list of nodes which are in the configuration.
1044 This function is for internal use, when the config lock is already
1050 return self._config_data.nodes.keys()
1052 @locking.ssynchronized(_config_lock, shared=1)
1053 def GetNodeList(self):
1054 """Return the list of nodes which are in the configuration.
1057 return self._UnlockedGetNodeList()
1059 def _UnlockedGetOnlineNodeList(self):
1060 """Return the list of nodes which are online.
1063 all_nodes = [self._UnlockedGetNodeInfo(node)
1064 for node in self._UnlockedGetNodeList()]
1065 return [node.name for node in all_nodes if not node.offline]
1067 @locking.ssynchronized(_config_lock, shared=1)
1068 def GetOnlineNodeList(self):
1069 """Return the list of nodes which are online.
1072 return self._UnlockedGetOnlineNodeList()
1074 @locking.ssynchronized(_config_lock, shared=1)
1075 def GetAllNodesInfo(self):
1076 """Get the configuration of all nodes.
1079 @return: dict of (node, node_info), where node_info is what
1080 would GetNodeInfo return for the node
1083 my_dict = dict([(node, self._UnlockedGetNodeInfo(node))
1084 for node in self._UnlockedGetNodeList()])
1087 def _UnlockedGetMasterCandidateStats(self, exceptions=None):
1088 """Get the number of current and maximum desired and possible candidates.
1090 @type exceptions: list
1091 @param exceptions: if passed, list of nodes that should be ignored
1093 @return: tuple of (current, desired and possible, possible)
1096 mc_now = mc_should = mc_max = 0
1097 for node in self._config_data.nodes.values():
1098 if exceptions and node.name in exceptions:
1100 if not (node.offline or node.drained):
1102 if node.master_candidate:
1104 mc_should = min(mc_max, self._config_data.cluster.candidate_pool_size)
1105 return (mc_now, mc_should, mc_max)
1107 @locking.ssynchronized(_config_lock, shared=1)
1108 def GetMasterCandidateStats(self, exceptions=None):
1109 """Get the number of current and maximum possible candidates.
1111 This is just a wrapper over L{_UnlockedGetMasterCandidateStats}.
1113 @type exceptions: list
1114 @param exceptions: if passed, list of nodes that should be ignored
1116 @return: tuple of (current, max)
1119 return self._UnlockedGetMasterCandidateStats(exceptions)
1121 @locking.ssynchronized(_config_lock)
1122 def MaintainCandidatePool(self, exceptions):
1123 """Try to grow the candidate pool to the desired size.
1125 @type exceptions: list
1126 @param exceptions: if passed, list of nodes that should be ignored
1128 @return: list with the adjusted nodes (L{objects.Node} instances)
1131 mc_now, mc_max, _ = self._UnlockedGetMasterCandidateStats(exceptions)
1134 node_list = self._config_data.nodes.keys()
1135 random.shuffle(node_list)
1136 for name in node_list:
1137 if mc_now >= mc_max:
1139 node = self._config_data.nodes[name]
1140 if (node.master_candidate or node.offline or node.drained or
1141 node.name in exceptions):
1143 mod_list.append(node)
1144 node.master_candidate = True
1147 if mc_now != mc_max:
1148 # this should not happen
1149 logging.warning("Warning: MaintainCandidatePool didn't manage to"
1150 " fill the candidate pool (%d/%d)", mc_now, mc_max)
1152 self._config_data.cluster.serial_no += 1
1157 def _BumpSerialNo(self):
1158 """Bump up the serial number of the config.
1161 self._config_data.serial_no += 1
1162 self._config_data.mtime = time.time()
1164 def _AllUUIDObjects(self):
1165 """Returns all objects with uuid attributes.
1168 return (self._config_data.instances.values() +
1169 self._config_data.nodes.values() +
1170 [self._config_data.cluster])
1172 def _OpenConfig(self):
1173 """Read the config data from disk.
1176 raw_data = utils.ReadFile(self._cfg_file)
1179 data = objects.ConfigData.FromDict(serializer.Load(raw_data))
1180 except Exception, err:
1181 raise errors.ConfigurationError(err)
1183 # Make sure the configuration has the right version
1184 _ValidateConfig(data)
1186 if (not hasattr(data, 'cluster') or
1187 not hasattr(data.cluster, 'rsahostkeypub')):
1188 raise errors.ConfigurationError("Incomplete configuration"
1189 " (missing cluster.rsahostkeypub)")
1191 # Upgrade configuration if needed
1192 data.UpgradeConfig()
1194 self._config_data = data
1195 # reset the last serial as -1 so that the next write will cause
1197 self._last_cluster_serial = -1
1199 # And finally run our (custom) config upgrade sequence
1200 self._UpgradeConfig()
1202 def _UpgradeConfig(self):
1203 """Run upgrade steps that cannot be done purely in the objects.
1205 This is because some data elements need uniqueness across the
1206 whole configuration, etc.
1208 @warning: this function will call L{_WriteConfig()}, so it needs
1209 to either be called with the lock held or from a safe place
1214 for item in self._AllUUIDObjects():
1215 if item.uuid is None:
1216 item.uuid = self._GenerateUniqueID(_UPGRADE_CONFIG_JID)
1220 # This is ok even if it acquires the internal lock, as _UpgradeConfig is
1221 # only called at config init time, without the lock held
1222 self.DropECReservations(_UPGRADE_CONFIG_JID)
1224 def _DistributeConfig(self, feedback_fn):
1225 """Distribute the configuration to the other nodes.
1227 Currently, this only copies the configuration file. In the future,
1228 it could be used to encapsulate the 2/3-phase update mechanism.
1238 myhostname = self._my_hostname
1239 # we can skip checking whether _UnlockedGetNodeInfo returns None
1240 # since the node list comes from _UnlocketGetNodeList, and we are
1241 # called with the lock held, so no modifications should take place
1243 for node_name in self._UnlockedGetNodeList():
1244 if node_name == myhostname:
1246 node_info = self._UnlockedGetNodeInfo(node_name)
1247 if not node_info.master_candidate:
1249 node_list.append(node_info.name)
1250 addr_list.append(node_info.primary_ip)
1252 result = rpc.RpcRunner.call_upload_file(node_list, self._cfg_file,
1253 address_list=addr_list)
1254 for to_node, to_result in result.items():
1255 msg = to_result.fail_msg
1257 msg = ("Copy of file %s to node %s failed: %s" %
1258 (self._cfg_file, to_node, msg))
1268 def _WriteConfig(self, destination=None, feedback_fn=None):
1269 """Write the configuration data to persistent storage.
1272 assert feedback_fn is None or callable(feedback_fn)
1274 # Warn on config errors, but don't abort the save - the
1275 # configuration has already been modified, and we can't revert;
1276 # the best we can do is to warn the user and save as is, leaving
1277 # recovery to the user
1278 config_errors = self._UnlockedVerifyConfig()
1280 errmsg = ("Configuration data is not consistent: %s" %
1281 (utils.CommaJoin(config_errors)))
1282 logging.critical(errmsg)
1286 if destination is None:
1287 destination = self._cfg_file
1288 self._BumpSerialNo()
1289 txt = serializer.Dump(self._config_data.ToDict())
1291 utils.WriteFile(destination, data=txt)
1293 self.write_count += 1
1295 # and redistribute the config file to master candidates
1296 self._DistributeConfig(feedback_fn)
1298 # Write ssconf files on all nodes (including locally)
1299 if self._last_cluster_serial < self._config_data.cluster.serial_no:
1300 if not self._offline:
1301 result = rpc.RpcRunner.call_write_ssconf_files(
1302 self._UnlockedGetOnlineNodeList(),
1303 self._UnlockedGetSsconfValues())
1305 for nname, nresu in result.items():
1306 msg = nresu.fail_msg
1308 errmsg = ("Error while uploading ssconf files to"
1309 " node %s: %s" % (nname, msg))
1310 logging.warning(errmsg)
1315 self._last_cluster_serial = self._config_data.cluster.serial_no
1317 def _UnlockedGetSsconfValues(self):
1318 """Return the values needed by ssconf.
1321 @return: a dictionary with keys the ssconf names and values their
1326 instance_names = utils.NiceSort(self._UnlockedGetInstanceList())
1327 node_names = utils.NiceSort(self._UnlockedGetNodeList())
1328 node_info = [self._UnlockedGetNodeInfo(name) for name in node_names]
1329 node_pri_ips = ["%s %s" % (ninfo.name, ninfo.primary_ip)
1330 for ninfo in node_info]
1331 node_snd_ips = ["%s %s" % (ninfo.name, ninfo.secondary_ip)
1332 for ninfo in node_info]
1334 instance_data = fn(instance_names)
1335 off_data = fn(node.name for node in node_info if node.offline)
1336 on_data = fn(node.name for node in node_info if not node.offline)
1337 mc_data = fn(node.name for node in node_info if node.master_candidate)
1338 mc_ips_data = fn(node.primary_ip for node in node_info
1339 if node.master_candidate)
1340 node_data = fn(node_names)
1341 node_pri_ips_data = fn(node_pri_ips)
1342 node_snd_ips_data = fn(node_snd_ips)
1344 cluster = self._config_data.cluster
1345 cluster_tags = fn(cluster.GetTags())
1347 constants.SS_CLUSTER_NAME: cluster.cluster_name,
1348 constants.SS_CLUSTER_TAGS: cluster_tags,
1349 constants.SS_FILE_STORAGE_DIR: cluster.file_storage_dir,
1350 constants.SS_MASTER_CANDIDATES: mc_data,
1351 constants.SS_MASTER_CANDIDATES_IPS: mc_ips_data,
1352 constants.SS_MASTER_IP: cluster.master_ip,
1353 constants.SS_MASTER_NETDEV: cluster.master_netdev,
1354 constants.SS_MASTER_NODE: cluster.master_node,
1355 constants.SS_NODE_LIST: node_data,
1356 constants.SS_NODE_PRIMARY_IPS: node_pri_ips_data,
1357 constants.SS_NODE_SECONDARY_IPS: node_snd_ips_data,
1358 constants.SS_OFFLINE_NODES: off_data,
1359 constants.SS_ONLINE_NODES: on_data,
1360 constants.SS_INSTANCE_LIST: instance_data,
1361 constants.SS_RELEASE_VERSION: constants.RELEASE_VERSION,
1364 @locking.ssynchronized(_config_lock, shared=1)
1365 def GetVGName(self):
1366 """Return the volume group name.
1369 return self._config_data.cluster.volume_group_name
1371 @locking.ssynchronized(_config_lock)
1372 def SetVGName(self, vg_name):
1373 """Set the volume group name.
1376 self._config_data.cluster.volume_group_name = vg_name
1377 self._config_data.cluster.serial_no += 1
1380 @locking.ssynchronized(_config_lock, shared=1)
1381 def GetMACPrefix(self):
1382 """Return the mac prefix.
1385 return self._config_data.cluster.mac_prefix
1387 @locking.ssynchronized(_config_lock, shared=1)
1388 def GetClusterInfo(self):
1389 """Returns information about the cluster
1391 @rtype: L{objects.Cluster}
1392 @return: the cluster object
1395 return self._config_data.cluster
1397 @locking.ssynchronized(_config_lock)
1398 def Update(self, target, feedback_fn):
1399 """Notify function to be called after updates.
1401 This function must be called when an object (as returned by
1402 GetInstanceInfo, GetNodeInfo, GetCluster) has been updated and the
1403 caller wants the modifications saved to the backing store. Note
1404 that all modified objects will be saved, but the target argument
1405 is the one the caller wants to ensure that it's saved.
1407 @param target: an instance of either L{objects.Cluster},
1408 L{objects.Node} or L{objects.Instance} which is existing in
1410 @param feedback_fn: Callable feedback function
1413 if self._config_data is None:
1414 raise errors.ProgrammerError("Configuration file not read,"
1416 update_serial = False
1417 if isinstance(target, objects.Cluster):
1418 test = target == self._config_data.cluster
1419 elif isinstance(target, objects.Node):
1420 test = target in self._config_data.nodes.values()
1421 update_serial = True
1422 elif isinstance(target, objects.Instance):
1423 test = target in self._config_data.instances.values()
1425 raise errors.ProgrammerError("Invalid object type (%s) passed to"
1426 " ConfigWriter.Update" % type(target))
1428 raise errors.ConfigurationError("Configuration updated since object"
1429 " has been read or unknown object")
1430 target.serial_no += 1
1431 target.mtime = now = time.time()
1434 # for node updates, we need to increase the cluster serial too
1435 self._config_data.cluster.serial_no += 1
1436 self._config_data.cluster.mtime = now
1438 if isinstance(target, objects.Instance):
1439 self._UnlockedReleaseDRBDMinors(target.name)
1441 self._WriteConfig(feedback_fn=feedback_fn)
1443 @locking.ssynchronized(_config_lock)
1444 def DropECReservations(self, ec_id):
1445 """Drop per-execution-context reservations
1448 self._temporary_ids.DropECReservations(ec_id)
1449 self._temporary_macs.DropECReservations(ec_id)
1450 self._temporary_secrets.DropECReservations(ec_id)