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")
151 def _ComputeAllLVs(self):
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 @locking.ssynchronized(_config_lock, shared=1)
163 def GenerateUniqueID(self, exceptions=None):
164 """Generate an unique disk name.
166 This checks the current node, instances and disk names for
169 @param exceptions: a list with some other names which should be checked
170 for uniqueness (used for example when you want to get
171 more than one id at one time without adding each one in
172 turn to the config file)
175 @return: the unique id
179 existing.update(self._temporary_ids)
180 existing.update(self._ComputeAllLVs())
181 existing.update(self._config_data.instances.keys())
182 existing.update(self._config_data.nodes.keys())
183 if exceptions is not None:
184 existing.update(exceptions)
187 unique_id = utils.NewUUID()
188 if unique_id not in existing and unique_id is not None:
191 raise errors.ConfigurationError("Not able generate an unique ID"
192 " (last tried ID: %s" % unique_id)
193 self._temporary_ids.add(unique_id)
197 """Return all MACs present in the config.
200 @return: the list of all MACs
204 for instance in self._config_data.instances.values():
205 for nic in instance.nics:
206 result.append(nic.mac)
210 def _AllDRBDSecrets(self):
211 """Return all DRBD secrets present in the config.
214 @return: the list of all DRBD secrets
217 def helper(disk, result):
218 """Recursively gather secrets from this disk."""
219 if disk.dev_type == constants.DT_DRBD8:
220 result.append(disk.logical_id[5])
222 for child in disk.children:
223 helper(child, result)
226 for instance in self._config_data.instances.values():
227 for disk in instance.disks:
232 def _CheckDiskIDs(self, disk, l_ids, p_ids):
233 """Compute duplicate disk IDs
235 @type disk: L{objects.Disk}
236 @param disk: the disk at which to start searching
238 @param l_ids: list of current logical ids
240 @param p_ids: list of current physical ids
242 @return: a list of error messages
246 if disk.logical_id is not None:
247 if disk.logical_id in l_ids:
248 result.append("duplicate logical id %s" % str(disk.logical_id))
250 l_ids.append(disk.logical_id)
251 if disk.physical_id is not None:
252 if disk.physical_id in p_ids:
253 result.append("duplicate physical id %s" % str(disk.physical_id))
255 p_ids.append(disk.physical_id)
258 for child in disk.children:
259 result.extend(self._CheckDiskIDs(child, l_ids, p_ids))
262 def _UnlockedVerifyConfig(self):
266 @return: a list of error messages; a non-empty list signifies
273 data = self._config_data
276 for instance_name in data.instances:
277 instance = data.instances[instance_name]
278 if instance.primary_node not in data.nodes:
279 result.append("instance '%s' has invalid primary node '%s'" %
280 (instance_name, instance.primary_node))
281 for snode in instance.secondary_nodes:
282 if snode not in data.nodes:
283 result.append("instance '%s' has invalid secondary node '%s'" %
284 (instance_name, snode))
285 for idx, nic in enumerate(instance.nics):
286 if nic.mac in seen_macs:
287 result.append("instance '%s' has NIC %d mac %s duplicate" %
288 (instance_name, idx, nic.mac))
290 seen_macs.append(nic.mac)
292 # gather the drbd ports for duplicate checks
293 for dsk in instance.disks:
294 if dsk.dev_type in constants.LDS_DRBD:
295 tcp_port = dsk.logical_id[2]
296 if tcp_port not in ports:
298 ports[tcp_port].append((instance.name, "drbd disk %s" % dsk.iv_name))
299 # gather network port reservation
300 net_port = getattr(instance, "network_port", None)
301 if net_port is not None:
302 if net_port not in ports:
304 ports[net_port].append((instance.name, "network port"))
306 # instance disk verify
307 for idx, disk in enumerate(instance.disks):
308 result.extend(["instance '%s' disk %d error: %s" %
309 (instance.name, idx, msg) for msg in disk.Verify()])
310 result.extend(self._CheckDiskIDs(disk, seen_lids, seen_pids))
312 # cluster-wide pool of free ports
313 for free_port in data.cluster.tcpudp_port_pool:
314 if free_port not in ports:
315 ports[free_port] = []
316 ports[free_port].append(("cluster", "port marked as free"))
318 # compute tcp/udp duplicate ports
324 txt = ", ".join(["%s/%s" % val for val in pdata])
325 result.append("tcp/udp port %s has duplicates: %s" % (pnum, txt))
327 # highest used tcp port check
329 if keys[-1] > data.cluster.highest_used_port:
330 result.append("Highest used port mismatch, saved %s, computed %s" %
331 (data.cluster.highest_used_port, keys[-1]))
333 if not data.nodes[data.cluster.master_node].master_candidate:
334 result.append("Master node is not a master candidate")
336 # master candidate checks
337 mc_now, mc_max = self._UnlockedGetMasterCandidateStats()
339 result.append("Not enough master candidates: actual %d, target %d" %
343 for node in data.nodes.values():
344 if [node.master_candidate, node.drained, node.offline].count(True) > 1:
345 result.append("Node %s state is invalid: master_candidate=%s,"
346 " drain=%s, offline=%s" %
347 (node.name, node.master_candidate, node.drain,
351 d_map, duplicates = self._UnlockedComputeDRBDMap()
352 for node, minor, instance_a, instance_b in duplicates:
353 result.append("DRBD minor %d on node %s is assigned twice to instances"
354 " %s and %s" % (minor, node, instance_a, instance_b))
358 @locking.ssynchronized(_config_lock, shared=1)
359 def VerifyConfig(self):
362 This is just a wrapper over L{_UnlockedVerifyConfig}.
365 @return: a list of error messages; a non-empty list signifies
369 return self._UnlockedVerifyConfig()
371 def _UnlockedSetDiskID(self, disk, node_name):
372 """Convert the unique ID to the ID needed on the target nodes.
374 This is used only for drbd, which needs ip/port configuration.
376 The routine descends down and updates its children also, because
377 this helps when the only the top device is passed to the remote
380 This function is for internal use, when the config lock is already held.
384 for child in disk.children:
385 self._UnlockedSetDiskID(child, node_name)
387 if disk.logical_id is None and disk.physical_id is not None:
389 if disk.dev_type == constants.LD_DRBD8:
390 pnode, snode, port, pminor, sminor, secret = disk.logical_id
391 if node_name not in (pnode, snode):
392 raise errors.ConfigurationError("DRBD device not knowing node %s" %
394 pnode_info = self._UnlockedGetNodeInfo(pnode)
395 snode_info = self._UnlockedGetNodeInfo(snode)
396 if pnode_info is None or snode_info is None:
397 raise errors.ConfigurationError("Can't find primary or secondary node"
398 " for %s" % str(disk))
399 p_data = (pnode_info.secondary_ip, port)
400 s_data = (snode_info.secondary_ip, port)
401 if pnode == node_name:
402 disk.physical_id = p_data + s_data + (pminor, secret)
403 else: # it must be secondary, we tested above
404 disk.physical_id = s_data + p_data + (sminor, secret)
406 disk.physical_id = disk.logical_id
409 @locking.ssynchronized(_config_lock)
410 def SetDiskID(self, disk, node_name):
411 """Convert the unique ID to the ID needed on the target nodes.
413 This is used only for drbd, which needs ip/port configuration.
415 The routine descends down and updates its children also, because
416 this helps when the only the top device is passed to the remote
420 return self._UnlockedSetDiskID(disk, node_name)
422 @locking.ssynchronized(_config_lock)
423 def AddTcpUdpPort(self, port):
424 """Adds a new port to the available port pool.
427 if not isinstance(port, int):
428 raise errors.ProgrammerError("Invalid type passed for port")
430 self._config_data.cluster.tcpudp_port_pool.add(port)
433 @locking.ssynchronized(_config_lock, shared=1)
434 def GetPortList(self):
435 """Returns a copy of the current port list.
438 return self._config_data.cluster.tcpudp_port_pool.copy()
440 @locking.ssynchronized(_config_lock)
441 def AllocatePort(self):
444 The port will be taken from the available port pool or from the
445 default port range (and in this case we increase
449 # If there are TCP/IP ports configured, we use them first.
450 if self._config_data.cluster.tcpudp_port_pool:
451 port = self._config_data.cluster.tcpudp_port_pool.pop()
453 port = self._config_data.cluster.highest_used_port + 1
454 if port >= constants.LAST_DRBD_PORT:
455 raise errors.ConfigurationError("The highest used port is greater"
456 " than %s. Aborting." %
457 constants.LAST_DRBD_PORT)
458 self._config_data.cluster.highest_used_port = port
463 def _UnlockedComputeDRBDMap(self):
464 """Compute the used DRBD minor/nodes.
467 @return: dictionary of node_name: dict of minor: instance_name;
468 the returned dict will have all the nodes in it (even if with
469 an empty list), and a list of duplicates; if the duplicates
470 list is not empty, the configuration is corrupted and its caller
471 should raise an exception
474 def _AppendUsedPorts(instance_name, disk, used):
476 if disk.dev_type == constants.LD_DRBD8 and len(disk.logical_id) >= 5:
477 node_a, node_b, _, minor_a, minor_b = disk.logical_id[:5]
478 for node, port in ((node_a, minor_a), (node_b, minor_b)):
479 assert node in used, ("Node '%s' of instance '%s' not found"
480 " in node list" % (node, instance_name))
481 if port in used[node]:
482 duplicates.append((node, port, instance_name, used[node][port]))
484 used[node][port] = instance_name
486 for child in disk.children:
487 duplicates.extend(_AppendUsedPorts(instance_name, child, used))
491 my_dict = dict((node, {}) for node in self._config_data.nodes)
492 for instance in self._config_data.instances.itervalues():
493 for disk in instance.disks:
494 duplicates.extend(_AppendUsedPorts(instance.name, disk, my_dict))
495 for (node, minor), instance in self._temporary_drbds.iteritems():
496 if minor in my_dict[node] and my_dict[node][minor] != instance:
497 duplicates.append((node, minor, instance, my_dict[node][minor]))
499 my_dict[node][minor] = instance
500 return my_dict, duplicates
502 @locking.ssynchronized(_config_lock)
503 def ComputeDRBDMap(self):
504 """Compute the used DRBD minor/nodes.
506 This is just a wrapper over L{_UnlockedComputeDRBDMap}.
508 @return: dictionary of node_name: dict of minor: instance_name;
509 the returned dict will have all the nodes in it (even if with
513 d_map, duplicates = self._UnlockedComputeDRBDMap()
515 raise errors.ConfigurationError("Duplicate DRBD ports detected: %s" %
519 @locking.ssynchronized(_config_lock)
520 def AllocateDRBDMinor(self, nodes, instance):
521 """Allocate a drbd minor.
523 The free minor will be automatically computed from the existing
524 devices. A node can be given multiple times in order to allocate
525 multiple minors. The result is the list of minors, in the same
526 order as the passed nodes.
528 @type instance: string
529 @param instance: the instance for which we allocate minors
532 assert isinstance(instance, basestring), \
533 "Invalid argument '%s' passed to AllocateDRBDMinor" % instance
535 d_map, duplicates = self._UnlockedComputeDRBDMap()
537 raise errors.ConfigurationError("Duplicate DRBD ports detected: %s" %
543 # no minors used, we can start at 0
546 self._temporary_drbds[(nname, 0)] = instance
550 ffree = utils.FirstFree(keys)
552 # return the next minor
553 # TODO: implement high-limit check
557 # double-check minor against current instances
558 assert minor not in d_map[nname], \
559 ("Attempt to reuse allocated DRBD minor %d on node %s,"
560 " already allocated to instance %s" %
561 (minor, nname, d_map[nname][minor]))
562 ndata[minor] = instance
563 # double-check minor against reservation
564 r_key = (nname, minor)
565 assert r_key not in self._temporary_drbds, \
566 ("Attempt to reuse reserved DRBD minor %d on node %s,"
567 " reserved for instance %s" %
568 (minor, nname, self._temporary_drbds[r_key]))
569 self._temporary_drbds[r_key] = instance
571 logging.debug("Request to allocate drbd minors, input: %s, returning %s",
575 def _UnlockedReleaseDRBDMinors(self, instance):
576 """Release temporary drbd minors allocated for a given instance.
578 @type instance: string
579 @param instance: the instance for which temporary minors should be
583 assert isinstance(instance, basestring), \
584 "Invalid argument passed to ReleaseDRBDMinors"
585 for key, name in self._temporary_drbds.items():
587 del self._temporary_drbds[key]
589 @locking.ssynchronized(_config_lock)
590 def ReleaseDRBDMinors(self, instance):
591 """Release temporary drbd minors allocated for a given instance.
593 This should be called on the error paths, on the success paths
594 it's automatically called by the ConfigWriter add and update
597 This function is just a wrapper over L{_UnlockedReleaseDRBDMinors}.
599 @type instance: string
600 @param instance: the instance for which temporary minors should be
604 self._UnlockedReleaseDRBDMinors(instance)
606 @locking.ssynchronized(_config_lock, shared=1)
607 def GetConfigVersion(self):
608 """Get the configuration version.
610 @return: Config version
613 return self._config_data.version
615 @locking.ssynchronized(_config_lock, shared=1)
616 def GetClusterName(self):
619 @return: Cluster name
622 return self._config_data.cluster.cluster_name
624 @locking.ssynchronized(_config_lock, shared=1)
625 def GetMasterNode(self):
626 """Get the hostname of the master node for this cluster.
628 @return: Master hostname
631 return self._config_data.cluster.master_node
633 @locking.ssynchronized(_config_lock, shared=1)
634 def GetMasterIP(self):
635 """Get the IP of the master node for this cluster.
640 return self._config_data.cluster.master_ip
642 @locking.ssynchronized(_config_lock, shared=1)
643 def GetMasterNetdev(self):
644 """Get the master network device for this cluster.
647 return self._config_data.cluster.master_netdev
649 @locking.ssynchronized(_config_lock, shared=1)
650 def GetFileStorageDir(self):
651 """Get the file storage dir for this cluster.
654 return self._config_data.cluster.file_storage_dir
656 @locking.ssynchronized(_config_lock, shared=1)
657 def GetHypervisorType(self):
658 """Get the hypervisor type for this cluster.
661 return self._config_data.cluster.enabled_hypervisors[0]
663 @locking.ssynchronized(_config_lock, shared=1)
664 def GetHostKey(self):
665 """Return the rsa hostkey from the config.
668 @return: the rsa hostkey
671 return self._config_data.cluster.rsahostkeypub
673 @locking.ssynchronized(_config_lock)
674 def AddInstance(self, instance):
675 """Add an instance to the config.
677 This should be used after creating a new instance.
679 @type instance: L{objects.Instance}
680 @param instance: the instance object
683 if not isinstance(instance, objects.Instance):
684 raise errors.ProgrammerError("Invalid type passed to AddInstance")
686 if instance.disk_template != constants.DT_DISKLESS:
687 all_lvs = instance.MapLVsByNode()
688 logging.info("Instance '%s' DISK_LAYOUT: %s", instance.name, all_lvs)
690 all_macs = self._AllMACs()
691 for nic in instance.nics:
692 if nic.mac in all_macs:
693 raise errors.ConfigurationError("Cannot add instance %s:"
694 " MAC address '%s' already in use." % (instance.name, nic.mac))
696 instance.serial_no = 1
697 self._config_data.instances[instance.name] = instance
698 self._config_data.cluster.serial_no += 1
699 self._UnlockedReleaseDRBDMinors(instance.name)
700 for nic in instance.nics:
701 self._temporary_macs.discard(nic.mac)
704 def _SetInstanceStatus(self, instance_name, status):
705 """Set the instance's status to a given value.
708 assert isinstance(status, bool), \
709 "Invalid status '%s' passed to SetInstanceStatus" % (status,)
711 if instance_name not in self._config_data.instances:
712 raise errors.ConfigurationError("Unknown instance '%s'" %
714 instance = self._config_data.instances[instance_name]
715 if instance.admin_up != status:
716 instance.admin_up = status
717 instance.serial_no += 1
720 @locking.ssynchronized(_config_lock)
721 def MarkInstanceUp(self, instance_name):
722 """Mark the instance status to up in the config.
725 self._SetInstanceStatus(instance_name, True)
727 @locking.ssynchronized(_config_lock)
728 def RemoveInstance(self, instance_name):
729 """Remove the instance from the configuration.
732 if instance_name not in self._config_data.instances:
733 raise errors.ConfigurationError("Unknown instance '%s'" % instance_name)
734 del self._config_data.instances[instance_name]
735 self._config_data.cluster.serial_no += 1
738 @locking.ssynchronized(_config_lock)
739 def RenameInstance(self, old_name, new_name):
740 """Rename an instance.
742 This needs to be done in ConfigWriter and not by RemoveInstance
743 combined with AddInstance as only we can guarantee an atomic
747 if old_name not in self._config_data.instances:
748 raise errors.ConfigurationError("Unknown instance '%s'" % old_name)
749 inst = self._config_data.instances[old_name]
750 del self._config_data.instances[old_name]
753 for disk in inst.disks:
754 if disk.dev_type == constants.LD_FILE:
755 # rename the file paths in logical and physical id
756 file_storage_dir = os.path.dirname(os.path.dirname(disk.logical_id[1]))
757 disk.physical_id = disk.logical_id = (disk.logical_id[0],
758 os.path.join(file_storage_dir,
762 self._config_data.instances[inst.name] = inst
765 @locking.ssynchronized(_config_lock)
766 def MarkInstanceDown(self, instance_name):
767 """Mark the status of an instance to down in the configuration.
770 self._SetInstanceStatus(instance_name, False)
772 def _UnlockedGetInstanceList(self):
773 """Get the list of instances.
775 This function is for internal use, when the config lock is already held.
778 return self._config_data.instances.keys()
780 @locking.ssynchronized(_config_lock, shared=1)
781 def GetInstanceList(self):
782 """Get the list of instances.
784 @return: array of instances, ex. ['instance2.example.com',
785 'instance1.example.com']
788 return self._UnlockedGetInstanceList()
790 @locking.ssynchronized(_config_lock, shared=1)
791 def ExpandInstanceName(self, short_name):
792 """Attempt to expand an incomplete instance name.
795 return utils.MatchNameComponent(short_name,
796 self._config_data.instances.keys())
798 def _UnlockedGetInstanceInfo(self, instance_name):
799 """Returns information about an instance.
801 This function is for internal use, when the config lock is already held.
804 if instance_name not in self._config_data.instances:
807 return self._config_data.instances[instance_name]
809 @locking.ssynchronized(_config_lock, shared=1)
810 def GetInstanceInfo(self, instance_name):
811 """Returns information about an instance.
813 It takes the information from the configuration file. Other information of
814 an instance are taken from the live systems.
816 @param instance_name: name of the instance, e.g.
817 I{instance1.example.com}
819 @rtype: L{objects.Instance}
820 @return: the instance object
823 return self._UnlockedGetInstanceInfo(instance_name)
825 @locking.ssynchronized(_config_lock, shared=1)
826 def GetAllInstancesInfo(self):
827 """Get the configuration of all instances.
830 @return: dict of (instance, instance_info), where instance_info is what
831 would GetInstanceInfo return for the node
834 my_dict = dict([(instance, self._UnlockedGetInstanceInfo(instance))
835 for instance in self._UnlockedGetInstanceList()])
838 @locking.ssynchronized(_config_lock)
839 def AddNode(self, node):
840 """Add a node to the configuration.
842 @type node: L{objects.Node}
843 @param node: a Node instance
846 logging.info("Adding node %s to configuration" % node.name)
849 self._config_data.nodes[node.name] = node
850 self._config_data.cluster.serial_no += 1
853 @locking.ssynchronized(_config_lock)
854 def RemoveNode(self, node_name):
855 """Remove a node from the configuration.
858 logging.info("Removing node %s from configuration" % node_name)
860 if node_name not in self._config_data.nodes:
861 raise errors.ConfigurationError("Unknown node '%s'" % node_name)
863 del self._config_data.nodes[node_name]
864 self._config_data.cluster.serial_no += 1
867 @locking.ssynchronized(_config_lock, shared=1)
868 def ExpandNodeName(self, short_name):
869 """Attempt to expand an incomplete instance name.
872 return utils.MatchNameComponent(short_name,
873 self._config_data.nodes.keys())
875 def _UnlockedGetNodeInfo(self, node_name):
876 """Get the configuration of a node, as stored in the config.
878 This function is for internal use, when the config lock is already
881 @param node_name: the node name, e.g. I{node1.example.com}
883 @rtype: L{objects.Node}
884 @return: the node object
887 if node_name not in self._config_data.nodes:
890 return self._config_data.nodes[node_name]
893 @locking.ssynchronized(_config_lock, shared=1)
894 def GetNodeInfo(self, node_name):
895 """Get the configuration of a node, as stored in the config.
897 This is just a locked wrapper over L{_UnlockedGetNodeInfo}.
899 @param node_name: the node name, e.g. I{node1.example.com}
901 @rtype: L{objects.Node}
902 @return: the node object
905 return self._UnlockedGetNodeInfo(node_name)
907 def _UnlockedGetNodeList(self):
908 """Return the list of nodes which are in the configuration.
910 This function is for internal use, when the config lock is already
916 return self._config_data.nodes.keys()
919 @locking.ssynchronized(_config_lock, shared=1)
920 def GetNodeList(self):
921 """Return the list of nodes which are in the configuration.
924 return self._UnlockedGetNodeList()
926 @locking.ssynchronized(_config_lock, shared=1)
927 def GetOnlineNodeList(self):
928 """Return the list of nodes which are online.
931 all_nodes = [self._UnlockedGetNodeInfo(node)
932 for node in self._UnlockedGetNodeList()]
933 return [node.name for node in all_nodes if not node.offline]
935 @locking.ssynchronized(_config_lock, shared=1)
936 def GetAllNodesInfo(self):
937 """Get the configuration of all nodes.
940 @return: dict of (node, node_info), where node_info is what
941 would GetNodeInfo return for the node
944 my_dict = dict([(node, self._UnlockedGetNodeInfo(node))
945 for node in self._UnlockedGetNodeList()])
948 def _UnlockedGetMasterCandidateStats(self, exceptions=None):
949 """Get the number of current and maximum desired and possible candidates.
951 @type exceptions: list
952 @param exceptions: if passed, list of nodes that should be ignored
954 @return: tuple of (current, desired and possible)
958 for node in self._config_data.nodes.values():
959 if exceptions and node.name in exceptions:
961 if not (node.offline or node.drained):
963 if node.master_candidate:
965 mc_max = min(mc_max, self._config_data.cluster.candidate_pool_size)
966 return (mc_now, mc_max)
968 @locking.ssynchronized(_config_lock, shared=1)
969 def GetMasterCandidateStats(self, exceptions=None):
970 """Get the number of current and maximum possible candidates.
972 This is just a wrapper over L{_UnlockedGetMasterCandidateStats}.
974 @type exceptions: list
975 @param exceptions: if passed, list of nodes that should be ignored
977 @return: tuple of (current, max)
980 return self._UnlockedGetMasterCandidateStats(exceptions)
982 @locking.ssynchronized(_config_lock)
983 def MaintainCandidatePool(self):
984 """Try to grow the candidate pool to the desired size.
987 @return: list with the adjusted nodes (L{objects.Node} instances)
990 mc_now, mc_max = self._UnlockedGetMasterCandidateStats()
993 node_list = self._config_data.nodes.keys()
994 random.shuffle(node_list)
995 for name in node_list:
998 node = self._config_data.nodes[name]
999 if node.master_candidate or node.offline or node.drained:
1001 mod_list.append(node)
1002 node.master_candidate = True
1005 if mc_now != mc_max:
1006 # this should not happen
1007 logging.warning("Warning: MaintainCandidatePool didn't manage to"
1008 " fill the candidate pool (%d/%d)", mc_now, mc_max)
1010 self._config_data.cluster.serial_no += 1
1015 def _BumpSerialNo(self):
1016 """Bump up the serial number of the config.
1019 self._config_data.serial_no += 1
1021 def _OpenConfig(self):
1022 """Read the config data from disk.
1025 f = open(self._cfg_file, 'r')
1028 data = objects.ConfigData.FromDict(serializer.Load(f.read()))
1029 except Exception, err:
1030 raise errors.ConfigurationError(err)
1034 # Make sure the configuration has the right version
1035 _ValidateConfig(data)
1037 if (not hasattr(data, 'cluster') or
1038 not hasattr(data.cluster, 'rsahostkeypub')):
1039 raise errors.ConfigurationError("Incomplete configuration"
1040 " (missing cluster.rsahostkeypub)")
1041 self._config_data = data
1042 # reset the last serial as -1 so that the next write will cause
1044 self._last_cluster_serial = -1
1046 def _DistributeConfig(self):
1047 """Distribute the configuration to the other nodes.
1049 Currently, this only copies the configuration file. In the future,
1050 it could be used to encapsulate the 2/3-phase update mechanism.
1059 myhostname = self._my_hostname
1060 # we can skip checking whether _UnlockedGetNodeInfo returns None
1061 # since the node list comes from _UnlocketGetNodeList, and we are
1062 # called with the lock held, so no modifications should take place
1064 for node_name in self._UnlockedGetNodeList():
1065 if node_name == myhostname:
1067 node_info = self._UnlockedGetNodeInfo(node_name)
1068 if not node_info.master_candidate:
1070 node_list.append(node_info.name)
1071 addr_list.append(node_info.primary_ip)
1073 result = rpc.RpcRunner.call_upload_file(node_list, self._cfg_file,
1074 address_list=addr_list)
1075 for to_node, to_result in result.items():
1076 msg = to_result.RemoteFailMsg()
1078 msg = ("Copy of file %s to node %s failed: %s" %
1079 (self._cfg_file, to_node, msg))
1084 def _WriteConfig(self, destination=None):
1085 """Write the configuration data to persistent storage.
1088 config_errors = self._UnlockedVerifyConfig()
1090 raise errors.ConfigurationError("Configuration data is not"
1092 (", ".join(config_errors)))
1093 if destination is None:
1094 destination = self._cfg_file
1095 self._BumpSerialNo()
1096 txt = serializer.Dump(self._config_data.ToDict())
1097 dir_name, file_name = os.path.split(destination)
1098 fd, name = tempfile.mkstemp('.newconfig', file_name, dir_name)
1099 f = os.fdopen(fd, 'w')
1102 os.fsync(f.fileno())
1105 # we don't need to do os.close(fd) as f.close() did it
1106 os.rename(name, destination)
1107 self.write_count += 1
1109 # and redistribute the config file to master candidates
1110 self._DistributeConfig()
1112 # Write ssconf files on all nodes (including locally)
1113 if self._last_cluster_serial < self._config_data.cluster.serial_no:
1114 if not self._offline:
1115 result = rpc.RpcRunner.call_write_ssconf_files(\
1116 self._UnlockedGetNodeList(),
1117 self._UnlockedGetSsconfValues())
1118 for nname, nresu in result.items():
1119 msg = nresu.RemoteFailMsg()
1121 logging.warning("Error while uploading ssconf files to"
1122 " node %s: %s", nname, msg)
1123 self._last_cluster_serial = self._config_data.cluster.serial_no
1125 def _UnlockedGetSsconfValues(self):
1126 """Return the values needed by ssconf.
1129 @return: a dictionary with keys the ssconf names and values their
1134 instance_names = utils.NiceSort(self._UnlockedGetInstanceList())
1135 node_names = utils.NiceSort(self._UnlockedGetNodeList())
1136 node_info = [self._UnlockedGetNodeInfo(name) for name in node_names]
1138 instance_data = fn(instance_names)
1139 off_data = fn(node.name for node in node_info if node.offline)
1140 on_data = fn(node.name for node in node_info if not node.offline)
1141 mc_data = fn(node.name for node in node_info if node.master_candidate)
1142 node_data = fn(node_names)
1144 cluster = self._config_data.cluster
1145 cluster_tags = fn(cluster.GetTags())
1147 constants.SS_CLUSTER_NAME: cluster.cluster_name,
1148 constants.SS_CLUSTER_TAGS: cluster_tags,
1149 constants.SS_FILE_STORAGE_DIR: cluster.file_storage_dir,
1150 constants.SS_MASTER_CANDIDATES: mc_data,
1151 constants.SS_MASTER_IP: cluster.master_ip,
1152 constants.SS_MASTER_NETDEV: cluster.master_netdev,
1153 constants.SS_MASTER_NODE: cluster.master_node,
1154 constants.SS_NODE_LIST: node_data,
1155 constants.SS_OFFLINE_NODES: off_data,
1156 constants.SS_ONLINE_NODES: on_data,
1157 constants.SS_INSTANCE_LIST: instance_data,
1158 constants.SS_RELEASE_VERSION: constants.RELEASE_VERSION,
1161 @locking.ssynchronized(_config_lock)
1162 def InitConfig(self, version, cluster_config, master_node_config):
1163 """Create the initial cluster configuration.
1165 It will contain the current node, which will also be the master
1166 node, and no instances.
1169 @param version: Configuration version
1170 @type cluster_config: objects.Cluster
1171 @param cluster_config: Cluster configuration
1172 @type master_node_config: objects.Node
1173 @param master_node_config: Master node configuration
1177 master_node_config.name: master_node_config,
1180 self._config_data = objects.ConfigData(version=version,
1181 cluster=cluster_config,
1187 @locking.ssynchronized(_config_lock, shared=1)
1188 def GetVGName(self):
1189 """Return the volume group name.
1192 return self._config_data.cluster.volume_group_name
1194 @locking.ssynchronized(_config_lock)
1195 def SetVGName(self, vg_name):
1196 """Set the volume group name.
1199 self._config_data.cluster.volume_group_name = vg_name
1200 self._config_data.cluster.serial_no += 1
1203 @locking.ssynchronized(_config_lock, shared=1)
1204 def GetMACPrefix(self):
1205 """Return the mac prefix.
1208 return self._config_data.cluster.mac_prefix
1210 @locking.ssynchronized(_config_lock, shared=1)
1211 def GetClusterInfo(self):
1212 """Returns information about the cluster
1214 @rtype: L{objects.Cluster}
1215 @return: the cluster object
1218 return self._config_data.cluster
1220 @locking.ssynchronized(_config_lock)
1221 def Update(self, target):
1222 """Notify function to be called after updates.
1224 This function must be called when an object (as returned by
1225 GetInstanceInfo, GetNodeInfo, GetCluster) has been updated and the
1226 caller wants the modifications saved to the backing store. Note
1227 that all modified objects will be saved, but the target argument
1228 is the one the caller wants to ensure that it's saved.
1230 @param target: an instance of either L{objects.Cluster},
1231 L{objects.Node} or L{objects.Instance} which is existing in
1235 if self._config_data is None:
1236 raise errors.ProgrammerError("Configuration file not read,"
1238 update_serial = False
1239 if isinstance(target, objects.Cluster):
1240 test = target == self._config_data.cluster
1241 elif isinstance(target, objects.Node):
1242 test = target in self._config_data.nodes.values()
1243 update_serial = True
1244 elif isinstance(target, objects.Instance):
1245 test = target in self._config_data.instances.values()
1247 raise errors.ProgrammerError("Invalid object type (%s) passed to"
1248 " ConfigWriter.Update" % type(target))
1250 raise errors.ConfigurationError("Configuration updated since object"
1251 " has been read or unknown object")
1252 target.serial_no += 1
1255 # for node updates, we need to increase the cluster serial too
1256 self._config_data.cluster.serial_no += 1
1258 if isinstance(target, objects.Instance):
1259 self._UnlockedReleaseDRBDMinors(target.name)
1260 for nic in target.nics:
1261 self._temporary_macs.discard(nic.mac)