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()
51 def _ValidateConfig(data):
52 """Verifies that a configuration objects looks valid.
54 This only verifies the version of the configuration.
56 @raise errors.ConfigurationError: if the version differs from what
60 if data.version != constants.CONFIG_VERSION:
61 raise errors.ConfigurationError("Cluster configuration version"
62 " mismatch, got %s instead of %s" %
64 constants.CONFIG_VERSION))
68 """The interface to the cluster configuration.
71 def __init__(self, cfg_file=None, offline=False):
73 self._lock = _config_lock
74 self._config_data = None
75 self._offline = offline
77 self._cfg_file = constants.CLUSTER_CONF_FILE
79 self._cfg_file = cfg_file
80 self._temporary_ids = set()
81 self._temporary_drbds = {}
82 self._temporary_macs = set()
83 # Note: in order to prevent errors when resolving our name in
84 # _DistributeConfig, we compute it here once and reuse it; it's
85 # better to raise an error before starting to modify the config
86 # file than after it was modified
87 self._my_hostname = utils.HostInfo().name
88 self._last_cluster_serial = -1
91 # this method needs to be static, so that we can call it on the class
94 """Check if the cluster is configured.
97 return os.path.exists(constants.CLUSTER_CONF_FILE)
99 @locking.ssynchronized(_config_lock, shared=1)
100 def GenerateMAC(self):
101 """Generate a MAC for an instance.
103 This should check the current instances for duplicates.
106 prefix = self._config_data.cluster.mac_prefix
107 all_macs = self._AllMACs()
110 byte1 = random.randrange(0, 256)
111 byte2 = random.randrange(0, 256)
112 byte3 = random.randrange(0, 256)
113 mac = "%s:%02x:%02x:%02x" % (prefix, byte1, byte2, byte3)
114 if mac not in all_macs and mac not in self._temporary_macs:
118 raise errors.ConfigurationError("Can't generate unique MAC")
119 self._temporary_macs.add(mac)
122 @locking.ssynchronized(_config_lock, shared=1)
123 def IsMacInUse(self, mac):
124 """Predicate: check if the specified MAC is in use in the Ganeti cluster.
126 This only checks instances managed by this cluster, it does not
127 check for potential collisions elsewhere.
130 all_macs = self._AllMACs()
131 return mac in all_macs or mac in self._temporary_macs
133 @locking.ssynchronized(_config_lock, shared=1)
134 def GenerateDRBDSecret(self):
135 """Generate a DRBD secret.
137 This checks the current disks for duplicates.
140 all_secrets = self._AllDRBDSecrets()
143 secret = utils.GenerateSecret()
144 if secret not in all_secrets:
148 raise errors.ConfigurationError("Can't generate unique DRBD secret")
152 """Compute the list of all LVs.
156 for instance in self._config_data.instances.values():
157 node_data = instance.MapLVsByNode()
158 for lv_list in node_data.values():
159 lvnames.update(lv_list)
162 def _AllIDs(self, include_temporary):
163 """Compute the list of all UUIDs and names we have.
165 @type include_temporary: boolean
166 @param include_temporary: whether to include the _temporary_ids set
168 @return: a set of IDs
172 if include_temporary:
173 existing.update(self._temporary_ids)
174 existing.update(self._AllLVs())
175 existing.update(self._config_data.instances.keys())
176 existing.update(self._config_data.nodes.keys())
179 @locking.ssynchronized(_config_lock, shared=1)
180 def GenerateUniqueID(self, exceptions=None):
181 """Generate an unique disk name.
183 This checks the current node, instances and disk names for
186 @param exceptions: a list with some other names which should be checked
187 for uniqueness (used for example when you want to get
188 more than one id at one time without adding each one in
189 turn to the config file)
192 @return: the unique id
195 existing = self._AllIDs(include_temporary=True)
196 if exceptions is not None:
197 existing.update(exceptions)
200 unique_id = utils.NewUUID()
201 if unique_id not in existing and unique_id is not None:
204 raise errors.ConfigurationError("Not able generate an unique ID"
205 " (last tried ID: %s" % unique_id)
206 self._temporary_ids.add(unique_id)
209 def _CleanupTemporaryIDs(self):
210 """Cleanups the _temporary_ids structure.
213 existing = self._AllIDs(include_temporary=False)
214 self._temporary_ids = self._temporary_ids - existing
217 """Return all MACs present in the config.
220 @return: the list of all MACs
224 for instance in self._config_data.instances.values():
225 for nic in instance.nics:
226 result.append(nic.mac)
230 def _AllDRBDSecrets(self):
231 """Return all DRBD secrets present in the config.
234 @return: the list of all DRBD secrets
237 def helper(disk, result):
238 """Recursively gather secrets from this disk."""
239 if disk.dev_type == constants.DT_DRBD8:
240 result.append(disk.logical_id[5])
242 for child in disk.children:
243 helper(child, result)
246 for instance in self._config_data.instances.values():
247 for disk in instance.disks:
252 def _CheckDiskIDs(self, disk, l_ids, p_ids):
253 """Compute duplicate disk IDs
255 @type disk: L{objects.Disk}
256 @param disk: the disk at which to start searching
258 @param l_ids: list of current logical ids
260 @param p_ids: list of current physical ids
262 @return: a list of error messages
266 if disk.logical_id is not None:
267 if disk.logical_id in l_ids:
268 result.append("duplicate logical id %s" % str(disk.logical_id))
270 l_ids.append(disk.logical_id)
271 if disk.physical_id is not None:
272 if disk.physical_id in p_ids:
273 result.append("duplicate physical id %s" % str(disk.physical_id))
275 p_ids.append(disk.physical_id)
278 for child in disk.children:
279 result.extend(self._CheckDiskIDs(child, l_ids, p_ids))
282 def _UnlockedVerifyConfig(self):
286 @return: a list of error messages; a non-empty list signifies
293 data = self._config_data
297 # global cluster checks
298 if not data.cluster.enabled_hypervisors:
299 result.append("enabled hypervisors list doesn't have any entries")
300 invalid_hvs = set(data.cluster.enabled_hypervisors) - constants.HYPER_TYPES
302 result.append("enabled hypervisors contains invalid entries: %s" %
305 if data.cluster.master_node not in data.nodes:
306 result.append("cluster has invalid primary node '%s'" %
307 data.cluster.master_node)
309 # per-instance checks
310 for instance_name in data.instances:
311 instance = data.instances[instance_name]
312 if instance.primary_node not in data.nodes:
313 result.append("instance '%s' has invalid primary node '%s'" %
314 (instance_name, instance.primary_node))
315 for snode in instance.secondary_nodes:
316 if snode not in data.nodes:
317 result.append("instance '%s' has invalid secondary node '%s'" %
318 (instance_name, snode))
319 for idx, nic in enumerate(instance.nics):
320 if nic.mac in seen_macs:
321 result.append("instance '%s' has NIC %d mac %s duplicate" %
322 (instance_name, idx, nic.mac))
324 seen_macs.append(nic.mac)
326 # gather the drbd ports for duplicate checks
327 for dsk in instance.disks:
328 if dsk.dev_type in constants.LDS_DRBD:
329 tcp_port = dsk.logical_id[2]
330 if tcp_port not in ports:
332 ports[tcp_port].append((instance.name, "drbd disk %s" % dsk.iv_name))
333 # gather network port reservation
334 net_port = getattr(instance, "network_port", None)
335 if net_port is not None:
336 if net_port not in ports:
338 ports[net_port].append((instance.name, "network port"))
340 # instance disk verify
341 for idx, disk in enumerate(instance.disks):
342 result.extend(["instance '%s' disk %d error: %s" %
343 (instance.name, idx, msg) for msg in disk.Verify()])
344 result.extend(self._CheckDiskIDs(disk, seen_lids, seen_pids))
346 # cluster-wide pool of free ports
347 for free_port in data.cluster.tcpudp_port_pool:
348 if free_port not in ports:
349 ports[free_port] = []
350 ports[free_port].append(("cluster", "port marked as free"))
352 # compute tcp/udp duplicate ports
358 txt = ", ".join(["%s/%s" % val for val in pdata])
359 result.append("tcp/udp port %s has duplicates: %s" % (pnum, txt))
361 # highest used tcp port check
363 if keys[-1] > data.cluster.highest_used_port:
364 result.append("Highest used port mismatch, saved %s, computed %s" %
365 (data.cluster.highest_used_port, keys[-1]))
367 if not data.nodes[data.cluster.master_node].master_candidate:
368 result.append("Master node is not a master candidate")
370 # master candidate checks
371 mc_now, mc_max = self._UnlockedGetMasterCandidateStats()
373 result.append("Not enough master candidates: actual %d, target %d" %
377 for node in data.nodes.values():
378 if [node.master_candidate, node.drained, node.offline].count(True) > 1:
379 result.append("Node %s state is invalid: master_candidate=%s,"
380 " drain=%s, offline=%s" %
381 (node.name, node.master_candidate, node.drain,
385 d_map, duplicates = self._UnlockedComputeDRBDMap()
386 for node, minor, instance_a, instance_b in duplicates:
387 result.append("DRBD minor %d on node %s is assigned twice to instances"
388 " %s and %s" % (minor, node, instance_a, instance_b))
392 @locking.ssynchronized(_config_lock, shared=1)
393 def VerifyConfig(self):
396 This is just a wrapper over L{_UnlockedVerifyConfig}.
399 @return: a list of error messages; a non-empty list signifies
403 return self._UnlockedVerifyConfig()
405 def _UnlockedSetDiskID(self, disk, node_name):
406 """Convert the unique ID to the ID needed on the target nodes.
408 This is used only for drbd, which needs ip/port configuration.
410 The routine descends down and updates its children also, because
411 this helps when the only the top device is passed to the remote
414 This function is for internal use, when the config lock is already held.
418 for child in disk.children:
419 self._UnlockedSetDiskID(child, node_name)
421 if disk.logical_id is None and disk.physical_id is not None:
423 if disk.dev_type == constants.LD_DRBD8:
424 pnode, snode, port, pminor, sminor, secret = disk.logical_id
425 if node_name not in (pnode, snode):
426 raise errors.ConfigurationError("DRBD device not knowing node %s" %
428 pnode_info = self._UnlockedGetNodeInfo(pnode)
429 snode_info = self._UnlockedGetNodeInfo(snode)
430 if pnode_info is None or snode_info is None:
431 raise errors.ConfigurationError("Can't find primary or secondary node"
432 " for %s" % str(disk))
433 p_data = (pnode_info.secondary_ip, port)
434 s_data = (snode_info.secondary_ip, port)
435 if pnode == node_name:
436 disk.physical_id = p_data + s_data + (pminor, secret)
437 else: # it must be secondary, we tested above
438 disk.physical_id = s_data + p_data + (sminor, secret)
440 disk.physical_id = disk.logical_id
443 @locking.ssynchronized(_config_lock)
444 def SetDiskID(self, disk, node_name):
445 """Convert the unique ID to the ID needed on the target nodes.
447 This is used only for drbd, which needs ip/port configuration.
449 The routine descends down and updates its children also, because
450 this helps when the only the top device is passed to the remote
454 return self._UnlockedSetDiskID(disk, node_name)
456 @locking.ssynchronized(_config_lock)
457 def AddTcpUdpPort(self, port):
458 """Adds a new port to the available port pool.
461 if not isinstance(port, int):
462 raise errors.ProgrammerError("Invalid type passed for port")
464 self._config_data.cluster.tcpudp_port_pool.add(port)
467 @locking.ssynchronized(_config_lock, shared=1)
468 def GetPortList(self):
469 """Returns a copy of the current port list.
472 return self._config_data.cluster.tcpudp_port_pool.copy()
474 @locking.ssynchronized(_config_lock)
475 def AllocatePort(self):
478 The port will be taken from the available port pool or from the
479 default port range (and in this case we increase
483 # If there are TCP/IP ports configured, we use them first.
484 if self._config_data.cluster.tcpudp_port_pool:
485 port = self._config_data.cluster.tcpudp_port_pool.pop()
487 port = self._config_data.cluster.highest_used_port + 1
488 if port >= constants.LAST_DRBD_PORT:
489 raise errors.ConfigurationError("The highest used port is greater"
490 " than %s. Aborting." %
491 constants.LAST_DRBD_PORT)
492 self._config_data.cluster.highest_used_port = port
497 def _UnlockedComputeDRBDMap(self):
498 """Compute the used DRBD minor/nodes.
501 @return: dictionary of node_name: dict of minor: instance_name;
502 the returned dict will have all the nodes in it (even if with
503 an empty list), and a list of duplicates; if the duplicates
504 list is not empty, the configuration is corrupted and its caller
505 should raise an exception
508 def _AppendUsedPorts(instance_name, disk, used):
510 if disk.dev_type == constants.LD_DRBD8 and len(disk.logical_id) >= 5:
511 node_a, node_b, _, minor_a, minor_b = disk.logical_id[:5]
512 for node, port in ((node_a, minor_a), (node_b, minor_b)):
513 assert node in used, ("Node '%s' of instance '%s' not found"
514 " in node list" % (node, instance_name))
515 if port in used[node]:
516 duplicates.append((node, port, instance_name, used[node][port]))
518 used[node][port] = instance_name
520 for child in disk.children:
521 duplicates.extend(_AppendUsedPorts(instance_name, child, used))
525 my_dict = dict((node, {}) for node in self._config_data.nodes)
526 for instance in self._config_data.instances.itervalues():
527 for disk in instance.disks:
528 duplicates.extend(_AppendUsedPorts(instance.name, disk, my_dict))
529 for (node, minor), instance in self._temporary_drbds.iteritems():
530 if minor in my_dict[node] and my_dict[node][minor] != instance:
531 duplicates.append((node, minor, instance, my_dict[node][minor]))
533 my_dict[node][minor] = instance
534 return my_dict, duplicates
536 @locking.ssynchronized(_config_lock)
537 def ComputeDRBDMap(self):
538 """Compute the used DRBD minor/nodes.
540 This is just a wrapper over L{_UnlockedComputeDRBDMap}.
542 @return: dictionary of node_name: dict of minor: instance_name;
543 the returned dict will have all the nodes in it (even if with
547 d_map, duplicates = self._UnlockedComputeDRBDMap()
549 raise errors.ConfigurationError("Duplicate DRBD ports detected: %s" %
553 @locking.ssynchronized(_config_lock)
554 def AllocateDRBDMinor(self, nodes, instance):
555 """Allocate a drbd minor.
557 The free minor will be automatically computed from the existing
558 devices. A node can be given multiple times in order to allocate
559 multiple minors. The result is the list of minors, in the same
560 order as the passed nodes.
562 @type instance: string
563 @param instance: the instance for which we allocate minors
566 assert isinstance(instance, basestring), \
567 "Invalid argument '%s' passed to AllocateDRBDMinor" % instance
569 d_map, duplicates = self._UnlockedComputeDRBDMap()
571 raise errors.ConfigurationError("Duplicate DRBD ports detected: %s" %
577 # no minors used, we can start at 0
580 self._temporary_drbds[(nname, 0)] = instance
584 ffree = utils.FirstFree(keys)
586 # return the next minor
587 # TODO: implement high-limit check
591 # double-check minor against current instances
592 assert minor not in d_map[nname], \
593 ("Attempt to reuse allocated DRBD minor %d on node %s,"
594 " already allocated to instance %s" %
595 (minor, nname, d_map[nname][minor]))
596 ndata[minor] = instance
597 # double-check minor against reservation
598 r_key = (nname, minor)
599 assert r_key not in self._temporary_drbds, \
600 ("Attempt to reuse reserved DRBD minor %d on node %s,"
601 " reserved for instance %s" %
602 (minor, nname, self._temporary_drbds[r_key]))
603 self._temporary_drbds[r_key] = instance
605 logging.debug("Request to allocate drbd minors, input: %s, returning %s",
609 def _UnlockedReleaseDRBDMinors(self, instance):
610 """Release temporary drbd minors allocated for a given instance.
612 @type instance: string
613 @param instance: the instance for which temporary minors should be
617 assert isinstance(instance, basestring), \
618 "Invalid argument passed to ReleaseDRBDMinors"
619 for key, name in self._temporary_drbds.items():
621 del self._temporary_drbds[key]
623 @locking.ssynchronized(_config_lock)
624 def ReleaseDRBDMinors(self, instance):
625 """Release temporary drbd minors allocated for a given instance.
627 This should be called on the error paths, on the success paths
628 it's automatically called by the ConfigWriter add and update
631 This function is just a wrapper over L{_UnlockedReleaseDRBDMinors}.
633 @type instance: string
634 @param instance: the instance for which temporary minors should be
638 self._UnlockedReleaseDRBDMinors(instance)
640 @locking.ssynchronized(_config_lock, shared=1)
641 def GetConfigVersion(self):
642 """Get the configuration version.
644 @return: Config version
647 return self._config_data.version
649 @locking.ssynchronized(_config_lock, shared=1)
650 def GetClusterName(self):
653 @return: Cluster name
656 return self._config_data.cluster.cluster_name
658 @locking.ssynchronized(_config_lock, shared=1)
659 def GetMasterNode(self):
660 """Get the hostname of the master node for this cluster.
662 @return: Master hostname
665 return self._config_data.cluster.master_node
667 @locking.ssynchronized(_config_lock, shared=1)
668 def GetMasterIP(self):
669 """Get the IP of the master node for this cluster.
674 return self._config_data.cluster.master_ip
676 @locking.ssynchronized(_config_lock, shared=1)
677 def GetMasterNetdev(self):
678 """Get the master network device for this cluster.
681 return self._config_data.cluster.master_netdev
683 @locking.ssynchronized(_config_lock, shared=1)
684 def GetFileStorageDir(self):
685 """Get the file storage dir for this cluster.
688 return self._config_data.cluster.file_storage_dir
690 @locking.ssynchronized(_config_lock, shared=1)
691 def GetHypervisorType(self):
692 """Get the hypervisor type for this cluster.
695 return self._config_data.cluster.default_hypervisor
697 @locking.ssynchronized(_config_lock, shared=1)
698 def GetHostKey(self):
699 """Return the rsa hostkey from the config.
702 @return: the rsa hostkey
705 return self._config_data.cluster.rsahostkeypub
707 @locking.ssynchronized(_config_lock)
708 def AddInstance(self, instance):
709 """Add an instance to the config.
711 This should be used after creating a new instance.
713 @type instance: L{objects.Instance}
714 @param instance: the instance object
717 if not isinstance(instance, objects.Instance):
718 raise errors.ProgrammerError("Invalid type passed to AddInstance")
720 if instance.disk_template != constants.DT_DISKLESS:
721 all_lvs = instance.MapLVsByNode()
722 logging.info("Instance '%s' DISK_LAYOUT: %s", instance.name, all_lvs)
724 all_macs = self._AllMACs()
725 for nic in instance.nics:
726 if nic.mac in all_macs:
727 raise errors.ConfigurationError("Cannot add instance %s:"
728 " MAC address '%s' already in use." % (instance.name, nic.mac))
730 instance.serial_no = 1
731 self._config_data.instances[instance.name] = instance
732 self._config_data.cluster.serial_no += 1
733 self._UnlockedReleaseDRBDMinors(instance.name)
734 for nic in instance.nics:
735 self._temporary_macs.discard(nic.mac)
738 def _SetInstanceStatus(self, instance_name, status):
739 """Set the instance's status to a given value.
742 assert isinstance(status, bool), \
743 "Invalid status '%s' passed to SetInstanceStatus" % (status,)
745 if instance_name not in self._config_data.instances:
746 raise errors.ConfigurationError("Unknown instance '%s'" %
748 instance = self._config_data.instances[instance_name]
749 if instance.admin_up != status:
750 instance.admin_up = status
751 instance.serial_no += 1
754 @locking.ssynchronized(_config_lock)
755 def MarkInstanceUp(self, instance_name):
756 """Mark the instance status to up in the config.
759 self._SetInstanceStatus(instance_name, True)
761 @locking.ssynchronized(_config_lock)
762 def RemoveInstance(self, instance_name):
763 """Remove the instance from the configuration.
766 if instance_name not in self._config_data.instances:
767 raise errors.ConfigurationError("Unknown instance '%s'" % instance_name)
768 del self._config_data.instances[instance_name]
769 self._config_data.cluster.serial_no += 1
772 @locking.ssynchronized(_config_lock)
773 def RenameInstance(self, old_name, new_name):
774 """Rename an instance.
776 This needs to be done in ConfigWriter and not by RemoveInstance
777 combined with AddInstance as only we can guarantee an atomic
781 if old_name not in self._config_data.instances:
782 raise errors.ConfigurationError("Unknown instance '%s'" % old_name)
783 inst = self._config_data.instances[old_name]
784 del self._config_data.instances[old_name]
787 for disk in inst.disks:
788 if disk.dev_type == constants.LD_FILE:
789 # rename the file paths in logical and physical id
790 file_storage_dir = os.path.dirname(os.path.dirname(disk.logical_id[1]))
791 disk.physical_id = disk.logical_id = (disk.logical_id[0],
792 os.path.join(file_storage_dir,
796 self._config_data.instances[inst.name] = inst
799 @locking.ssynchronized(_config_lock)
800 def MarkInstanceDown(self, instance_name):
801 """Mark the status of an instance to down in the configuration.
804 self._SetInstanceStatus(instance_name, False)
806 def _UnlockedGetInstanceList(self):
807 """Get the list of instances.
809 This function is for internal use, when the config lock is already held.
812 return self._config_data.instances.keys()
814 @locking.ssynchronized(_config_lock, shared=1)
815 def GetInstanceList(self):
816 """Get the list of instances.
818 @return: array of instances, ex. ['instance2.example.com',
819 'instance1.example.com']
822 return self._UnlockedGetInstanceList()
824 @locking.ssynchronized(_config_lock, shared=1)
825 def ExpandInstanceName(self, short_name):
826 """Attempt to expand an incomplete instance name.
829 return utils.MatchNameComponent(short_name,
830 self._config_data.instances.keys())
832 def _UnlockedGetInstanceInfo(self, instance_name):
833 """Returns information about an instance.
835 This function is for internal use, when the config lock is already held.
838 if instance_name not in self._config_data.instances:
841 return self._config_data.instances[instance_name]
843 @locking.ssynchronized(_config_lock, shared=1)
844 def GetInstanceInfo(self, instance_name):
845 """Returns information about an instance.
847 It takes the information from the configuration file. Other information of
848 an instance are taken from the live systems.
850 @param instance_name: name of the instance, e.g.
851 I{instance1.example.com}
853 @rtype: L{objects.Instance}
854 @return: the instance object
857 return self._UnlockedGetInstanceInfo(instance_name)
859 @locking.ssynchronized(_config_lock, shared=1)
860 def GetAllInstancesInfo(self):
861 """Get the configuration of all instances.
864 @return: dict of (instance, instance_info), where instance_info is what
865 would GetInstanceInfo return for the node
868 my_dict = dict([(instance, self._UnlockedGetInstanceInfo(instance))
869 for instance in self._UnlockedGetInstanceList()])
872 @locking.ssynchronized(_config_lock)
873 def AddNode(self, node):
874 """Add a node to the configuration.
876 @type node: L{objects.Node}
877 @param node: a Node instance
880 logging.info("Adding node %s to configuration" % node.name)
883 self._config_data.nodes[node.name] = node
884 self._config_data.cluster.serial_no += 1
887 @locking.ssynchronized(_config_lock)
888 def RemoveNode(self, node_name):
889 """Remove a node from the configuration.
892 logging.info("Removing node %s from configuration" % node_name)
894 if node_name not in self._config_data.nodes:
895 raise errors.ConfigurationError("Unknown node '%s'" % node_name)
897 del self._config_data.nodes[node_name]
898 self._config_data.cluster.serial_no += 1
901 @locking.ssynchronized(_config_lock, shared=1)
902 def ExpandNodeName(self, short_name):
903 """Attempt to expand an incomplete instance name.
906 return utils.MatchNameComponent(short_name,
907 self._config_data.nodes.keys())
909 def _UnlockedGetNodeInfo(self, node_name):
910 """Get the configuration of a node, as stored in the config.
912 This function is for internal use, when the config lock is already
915 @param node_name: the node name, e.g. I{node1.example.com}
917 @rtype: L{objects.Node}
918 @return: the node object
921 if node_name not in self._config_data.nodes:
924 return self._config_data.nodes[node_name]
927 @locking.ssynchronized(_config_lock, shared=1)
928 def GetNodeInfo(self, node_name):
929 """Get the configuration of a node, as stored in the config.
931 This is just a locked wrapper over L{_UnlockedGetNodeInfo}.
933 @param node_name: the node name, e.g. I{node1.example.com}
935 @rtype: L{objects.Node}
936 @return: the node object
939 return self._UnlockedGetNodeInfo(node_name)
941 def _UnlockedGetNodeList(self):
942 """Return the list of nodes which are in the configuration.
944 This function is for internal use, when the config lock is already
950 return self._config_data.nodes.keys()
953 @locking.ssynchronized(_config_lock, shared=1)
954 def GetNodeList(self):
955 """Return the list of nodes which are in the configuration.
958 return self._UnlockedGetNodeList()
960 @locking.ssynchronized(_config_lock, shared=1)
961 def GetOnlineNodeList(self):
962 """Return the list of nodes which are online.
965 all_nodes = [self._UnlockedGetNodeInfo(node)
966 for node in self._UnlockedGetNodeList()]
967 return [node.name for node in all_nodes if not node.offline]
969 @locking.ssynchronized(_config_lock, shared=1)
970 def GetAllNodesInfo(self):
971 """Get the configuration of all nodes.
974 @return: dict of (node, node_info), where node_info is what
975 would GetNodeInfo return for the node
978 my_dict = dict([(node, self._UnlockedGetNodeInfo(node))
979 for node in self._UnlockedGetNodeList()])
982 def _UnlockedGetMasterCandidateStats(self, exceptions=None):
983 """Get the number of current and maximum desired and possible candidates.
985 @type exceptions: list
986 @param exceptions: if passed, list of nodes that should be ignored
988 @return: tuple of (current, desired and possible)
992 for node in self._config_data.nodes.values():
993 if exceptions and node.name in exceptions:
995 if not (node.offline or node.drained):
997 if node.master_candidate:
999 mc_max = min(mc_max, self._config_data.cluster.candidate_pool_size)
1000 return (mc_now, mc_max)
1002 @locking.ssynchronized(_config_lock, shared=1)
1003 def GetMasterCandidateStats(self, exceptions=None):
1004 """Get the number of current and maximum possible candidates.
1006 This is just a wrapper over L{_UnlockedGetMasterCandidateStats}.
1008 @type exceptions: list
1009 @param exceptions: if passed, list of nodes that should be ignored
1011 @return: tuple of (current, max)
1014 return self._UnlockedGetMasterCandidateStats(exceptions)
1016 @locking.ssynchronized(_config_lock)
1017 def MaintainCandidatePool(self):
1018 """Try to grow the candidate pool to the desired size.
1021 @return: list with the adjusted nodes (L{objects.Node} instances)
1024 mc_now, mc_max = self._UnlockedGetMasterCandidateStats()
1027 node_list = self._config_data.nodes.keys()
1028 random.shuffle(node_list)
1029 for name in node_list:
1030 if mc_now >= mc_max:
1032 node = self._config_data.nodes[name]
1033 if node.master_candidate or node.offline or node.drained:
1035 mod_list.append(node)
1036 node.master_candidate = True
1039 if mc_now != mc_max:
1040 # this should not happen
1041 logging.warning("Warning: MaintainCandidatePool didn't manage to"
1042 " fill the candidate pool (%d/%d)", mc_now, mc_max)
1044 self._config_data.cluster.serial_no += 1
1049 def _BumpSerialNo(self):
1050 """Bump up the serial number of the config.
1053 self._config_data.serial_no += 1
1055 def _OpenConfig(self):
1056 """Read the config data from disk.
1059 f = open(self._cfg_file, 'r')
1062 data = objects.ConfigData.FromDict(serializer.Load(f.read()))
1063 except Exception, err:
1064 raise errors.ConfigurationError(err)
1068 # Make sure the configuration has the right version
1069 _ValidateConfig(data)
1071 if (not hasattr(data, 'cluster') or
1072 not hasattr(data.cluster, 'rsahostkeypub')):
1073 raise errors.ConfigurationError("Incomplete configuration"
1074 " (missing cluster.rsahostkeypub)")
1076 # Upgrade configuration if needed
1077 data.UpgradeConfig()
1079 self._config_data = data
1080 # reset the last serial as -1 so that the next write will cause
1082 self._last_cluster_serial = -1
1084 def _DistributeConfig(self):
1085 """Distribute the configuration to the other nodes.
1087 Currently, this only copies the configuration file. In the future,
1088 it could be used to encapsulate the 2/3-phase update mechanism.
1097 myhostname = self._my_hostname
1098 # we can skip checking whether _UnlockedGetNodeInfo returns None
1099 # since the node list comes from _UnlocketGetNodeList, and we are
1100 # called with the lock held, so no modifications should take place
1102 for node_name in self._UnlockedGetNodeList():
1103 if node_name == myhostname:
1105 node_info = self._UnlockedGetNodeInfo(node_name)
1106 if not node_info.master_candidate:
1108 node_list.append(node_info.name)
1109 addr_list.append(node_info.primary_ip)
1111 result = rpc.RpcRunner.call_upload_file(node_list, self._cfg_file,
1112 address_list=addr_list)
1113 for node in node_list:
1114 if not result[node]:
1115 logging.error("copy of file %s to node %s failed",
1116 self._cfg_file, node)
1120 def _WriteConfig(self, destination=None):
1121 """Write the configuration data to persistent storage.
1124 # first, cleanup the _temporary_ids set, if an ID is now in the
1125 # other objects it should be discarded to prevent unbounded growth
1127 self._CleanupTemporaryIDs()
1128 config_errors = self._UnlockedVerifyConfig()
1130 raise errors.ConfigurationError("Configuration data is not"
1132 (", ".join(config_errors)))
1133 if destination is None:
1134 destination = self._cfg_file
1135 self._BumpSerialNo()
1136 txt = serializer.Dump(self._config_data.ToDict())
1137 dir_name, file_name = os.path.split(destination)
1138 fd, name = tempfile.mkstemp('.newconfig', file_name, dir_name)
1139 f = os.fdopen(fd, 'w')
1142 os.fsync(f.fileno())
1145 # we don't need to do os.close(fd) as f.close() did it
1146 os.rename(name, destination)
1147 self.write_count += 1
1149 # and redistribute the config file to master candidates
1150 self._DistributeConfig()
1152 # Write ssconf files on all nodes (including locally)
1153 if self._last_cluster_serial < self._config_data.cluster.serial_no:
1154 if not self._offline:
1155 rpc.RpcRunner.call_write_ssconf_files(self._UnlockedGetNodeList(),
1156 self._UnlockedGetSsconfValues())
1157 self._last_cluster_serial = self._config_data.cluster.serial_no
1159 def _UnlockedGetSsconfValues(self):
1160 """Return the values needed by ssconf.
1163 @return: a dictionary with keys the ssconf names and values their
1168 instance_names = utils.NiceSort(self._UnlockedGetInstanceList())
1169 node_names = utils.NiceSort(self._UnlockedGetNodeList())
1170 node_info = [self._UnlockedGetNodeInfo(name) for name in node_names]
1172 instance_data = fn(instance_names)
1173 off_data = fn(node.name for node in node_info if node.offline)
1174 on_data = fn(node.name for node in node_info if not node.offline)
1175 mc_data = fn(node.name for node in node_info if node.master_candidate)
1176 node_data = fn(node_names)
1178 cluster = self._config_data.cluster
1179 cluster_tags = fn(cluster.GetTags())
1181 constants.SS_CLUSTER_NAME: cluster.cluster_name,
1182 constants.SS_CLUSTER_TAGS: cluster_tags,
1183 constants.SS_FILE_STORAGE_DIR: cluster.file_storage_dir,
1184 constants.SS_MASTER_CANDIDATES: mc_data,
1185 constants.SS_MASTER_IP: cluster.master_ip,
1186 constants.SS_MASTER_NETDEV: cluster.master_netdev,
1187 constants.SS_MASTER_NODE: cluster.master_node,
1188 constants.SS_NODE_LIST: node_data,
1189 constants.SS_OFFLINE_NODES: off_data,
1190 constants.SS_ONLINE_NODES: on_data,
1191 constants.SS_INSTANCE_LIST: instance_data,
1192 constants.SS_RELEASE_VERSION: constants.RELEASE_VERSION,
1195 @locking.ssynchronized(_config_lock, shared=1)
1196 def GetVGName(self):
1197 """Return the volume group name.
1200 return self._config_data.cluster.volume_group_name
1202 @locking.ssynchronized(_config_lock)
1203 def SetVGName(self, vg_name):
1204 """Set the volume group name.
1207 self._config_data.cluster.volume_group_name = vg_name
1208 self._config_data.cluster.serial_no += 1
1211 @locking.ssynchronized(_config_lock, shared=1)
1212 def GetDefBridge(self):
1213 """Return the default bridge.
1216 return self._config_data.cluster.default_bridge
1218 @locking.ssynchronized(_config_lock, shared=1)
1219 def GetMACPrefix(self):
1220 """Return the mac prefix.
1223 return self._config_data.cluster.mac_prefix
1225 @locking.ssynchronized(_config_lock, shared=1)
1226 def GetClusterInfo(self):
1227 """Returns information about the cluster
1229 @rtype: L{objects.Cluster}
1230 @return: the cluster object
1233 return self._config_data.cluster
1235 @locking.ssynchronized(_config_lock)
1236 def Update(self, target):
1237 """Notify function to be called after updates.
1239 This function must be called when an object (as returned by
1240 GetInstanceInfo, GetNodeInfo, GetCluster) has been updated and the
1241 caller wants the modifications saved to the backing store. Note
1242 that all modified objects will be saved, but the target argument
1243 is the one the caller wants to ensure that it's saved.
1245 @param target: an instance of either L{objects.Cluster},
1246 L{objects.Node} or L{objects.Instance} which is existing in
1250 if self._config_data is None:
1251 raise errors.ProgrammerError("Configuration file not read,"
1253 update_serial = False
1254 if isinstance(target, objects.Cluster):
1255 test = target == self._config_data.cluster
1256 elif isinstance(target, objects.Node):
1257 test = target in self._config_data.nodes.values()
1258 update_serial = True
1259 elif isinstance(target, objects.Instance):
1260 test = target in self._config_data.instances.values()
1262 raise errors.ProgrammerError("Invalid object type (%s) passed to"
1263 " ConfigWriter.Update" % type(target))
1265 raise errors.ConfigurationError("Configuration updated since object"
1266 " has been read or unknown object")
1267 target.serial_no += 1
1270 # for node updates, we need to increase the cluster serial too
1271 self._config_data.cluster.serial_no += 1
1273 if isinstance(target, objects.Instance):
1274 self._UnlockedReleaseDRBDMinors(target.name)
1275 for nic in target.nics:
1276 self._temporary_macs.discard(nic.mac)