from ganeti import rpc
from ganeti import objects
from ganeti import serializer
-from ganeti import ssconf
_config_lock = locking.SharedLock()
-def ValidateConfig():
- sstore = ssconf.SimpleStore()
-
- if sstore.GetConfigVersion() != constants.CONFIG_VERSION:
+def _ValidateConfig(data):
+ if data.version != constants.CONFIG_VERSION:
raise errors.ConfigurationError("Cluster configuration version"
" mismatch, got %s instead of %s" %
- (sstore.GetConfigVersion(),
+ (data.version,
constants.CONFIG_VERSION))
self.write_count = 0
self._lock = _config_lock
self._config_data = None
- self._config_time = None
- self._config_size = None
- self._config_inode = None
self._offline = offline
if cfg_file is None:
self._cfg_file = constants.CLUSTER_CONF_FILE
# better to raise an error before starting to modify the config
# file than after it was modified
self._my_hostname = utils.HostInfo().name
+ self._OpenConfig()
# this method needs to be static, so that we can call it on the class
@staticmethod
This should check the current instances for duplicates.
"""
- self._OpenConfig()
prefix = self._config_data.cluster.mac_prefix
all_macs = self._AllMACs()
retries = 64
check for potential collisions elsewhere.
"""
- self._OpenConfig()
all_macs = self._AllMACs()
return mac in all_macs
This checks the current disks for duplicates.
"""
- self._OpenConfig()
all_secrets = self._AllDRBDSecrets()
retries = 64
while retries > 0:
"""Compute the list of all LVs.
"""
- self._OpenConfig()
lvnames = set()
for instance in self._config_data.instances.values():
node_data = instance.MapLVsByNode()
"""Return all MACs present in the config.
"""
- self._OpenConfig()
-
result = []
for instance in self._config_data.instances.values():
for nic in instance.nics:
@locking.ssynchronized(_config_lock, shared=1)
def VerifyConfig(self):
- """Stub verify function.
- """
- self._OpenConfig()
+ """Verify function.
+ """
result = []
seen_macs = []
ports = {}
ports[net_port].append((instance.name, "network port"))
# cluster-wide pool of free ports
- for free_port in self._config_data.cluster.tcpudp_port_pool:
+ for free_port in data.cluster.tcpudp_port_pool:
if free_port not in ports:
ports[free_port] = []
ports[free_port].append(("cluster", "port marked as free"))
# highest used tcp port check
if keys:
- if keys[-1] > self._config_data.cluster.highest_used_port:
+ if keys[-1] > data.cluster.highest_used_port:
result.append("Highest used port mismatch, saved %s, computed %s" %
- (self._config_data.cluster.highest_used_port,
- keys[-1]))
+ (data.cluster.highest_used_port, keys[-1]))
+
+ if not data.nodes[data.cluster.master_node].master_candidate:
+ result.append("Master node is not a master candidate")
+
+ cp_size = data.cluster.candidate_pool_size
+ num_c = 0
+ for node in data.nodes.values():
+ if node.master_candidate:
+ num_c += 1
+ if cp_size > num_c and num_c < len(data.nodes):
+ result.append("Not enough master candidates: actual %d, desired %d,"
+ " %d total nodes" % (num_c, cp_size, len(data.nodes)))
return result
if not isinstance(port, int):
raise errors.ProgrammerError("Invalid type passed for port")
- self._OpenConfig()
self._config_data.cluster.tcpudp_port_pool.add(port)
self._WriteConfig()
"""Returns a copy of the current port list.
"""
- self._OpenConfig()
return self._config_data.cluster.tcpudp_port_pool.copy()
@locking.ssynchronized(_config_lock)
highest_used_port).
"""
- self._OpenConfig()
-
# If there are TCP/IP ports configured, we use them first.
if self._config_data.cluster.tcpudp_port_pool:
port = self._config_data.cluster.tcpudp_port_pool.pop()
order as the passed nodes.
"""
- self._OpenConfig()
-
d_map = self._ComputeDRBDMap(instance)
result = []
for nname in nodes:
del self._temporary_drbds[key]
@locking.ssynchronized(_config_lock, shared=1)
+ def GetConfigVersion(self):
+ """Get the configuration version.
+
+ @return: Config version
+
+ """
+ return self._config_data.version
+
+ @locking.ssynchronized(_config_lock, shared=1)
+ def GetClusterName(self):
+ """Get cluster name.
+
+ @return: Cluster name
+
+ """
+ return self._config_data.cluster.cluster_name
+
+ @locking.ssynchronized(_config_lock, shared=1)
+ def GetMasterNode(self):
+ """Get the hostname of the master node for this cluster.
+
+ @return: Master hostname
+
+ """
+ return self._config_data.cluster.master_node
+
+ @locking.ssynchronized(_config_lock, shared=1)
+ def GetMasterIP(self):
+ """Get the IP of the master node for this cluster.
+
+ @return: Master IP
+
+ """
+ return self._config_data.cluster.master_ip
+
+ @locking.ssynchronized(_config_lock, shared=1)
+ def GetMasterNetdev(self):
+ """Get the master network device for this cluster.
+
+ """
+ return self._config_data.cluster.master_netdev
+
+ @locking.ssynchronized(_config_lock, shared=1)
+ def GetFileStorageDir(self):
+ """Get the file storage dir for this cluster.
+
+ """
+ return self._config_data.cluster.file_storage_dir
+
+ @locking.ssynchronized(_config_lock, shared=1)
+ def GetHypervisorType(self):
+ """Get the hypervisor type for this cluster.
+
+ """
+ return self._config_data.cluster.default_hypervisor
+
+ @locking.ssynchronized(_config_lock, shared=1)
def GetHostKey(self):
"""Return the rsa hostkey from the config.
Returns: rsa hostkey
"""
- self._OpenConfig()
return self._config_data.cluster.rsahostkeypub
@locking.ssynchronized(_config_lock)
all_lvs = instance.MapLVsByNode()
logging.info("Instance '%s' DISK_LAYOUT: %s", instance.name, all_lvs)
- self._OpenConfig()
instance.serial_no = 1
self._config_data.instances[instance.name] = instance
- self._config_data.cluster.serial_no += 1
self._WriteConfig()
def _SetInstanceStatus(self, instance_name, status):
raise errors.ProgrammerError("Invalid status '%s' passed to"
" ConfigWriter._SetInstanceStatus()" %
status)
- self._OpenConfig()
if instance_name not in self._config_data.instances:
raise errors.ConfigurationError("Unknown instance '%s'" %
"""Remove the instance from the configuration.
"""
- self._OpenConfig()
-
if instance_name not in self._config_data.instances:
raise errors.ConfigurationError("Unknown instance '%s'" % instance_name)
del self._config_data.instances[instance_name]
- self._config_data.cluster.serial_no += 1
self._WriteConfig()
@locking.ssynchronized(_config_lock)
rename.
"""
- self._OpenConfig()
if old_name not in self._config_data.instances:
raise errors.ConfigurationError("Unknown instance '%s'" % old_name)
inst = self._config_data.instances[old_name]
disk.iv_name))
self._config_data.instances[inst.name] = inst
- self._config_data.cluster.serial_no += 1
self._WriteConfig()
@locking.ssynchronized(_config_lock)
This function is for internal use, when the config lock is already held.
"""
- self._OpenConfig()
return self._config_data.instances.keys()
@locking.ssynchronized(_config_lock, shared=1)
"""Attempt to expand an incomplete instance name.
"""
- self._OpenConfig()
-
return utils.MatchNameComponent(short_name,
self._config_data.instances.keys())
This function is for internal use, when the config lock is already held.
"""
- self._OpenConfig()
-
if instance_name not in self._config_data.instances:
return None
"""
logging.info("Adding node %s to configuration" % node.name)
- self._OpenConfig()
node.serial_no = 1
self._config_data.nodes[node.name] = node
self._config_data.cluster.serial_no += 1
"""
logging.info("Removing node %s from configuration" % node_name)
- self._OpenConfig()
if node_name not in self._config_data.nodes:
raise errors.ConfigurationError("Unknown node '%s'" % node_name)
"""Attempt to expand an incomplete instance name.
"""
- self._OpenConfig()
-
return utils.MatchNameComponent(short_name,
self._config_data.nodes.keys())
Returns: the node object
"""
- self._OpenConfig()
-
if node_name not in self._config_data.nodes:
return None
This function is for internal use, when the config lock is already held.
"""
- self._OpenConfig()
return self._config_data.nodes.keys()
for node in self._UnlockedGetNodeList()])
return my_dict
- @locking.ssynchronized(_config_lock, shared=1)
- def DumpConfig(self):
- """Return the entire configuration of the cluster.
- """
- self._OpenConfig()
- return self._config_data
-
def _BumpSerialNo(self):
"""Bump up the serial number of the config.
file, since de-serialisation could be slow.
"""
- try:
- st = os.stat(self._cfg_file)
- except OSError, err:
- raise errors.ConfigurationError("Can't stat config file: %s" % err)
- if (self._config_data is not None and
- self._config_time is not None and
- self._config_time == st.st_mtime and
- self._config_size == st.st_size and
- self._config_inode == st.st_ino):
- # data is current, so skip loading of config file
- return
-
- # Make sure the configuration has the right version
- ValidateConfig()
-
f = open(self._cfg_file, 'r')
try:
try:
raise errors.ConfigurationError(err)
finally:
f.close()
+
+ # Make sure the configuration has the right version
+ _ValidateConfig(data)
+
if (not hasattr(data, 'cluster') or
not hasattr(data.cluster, 'rsahostkeypub')):
raise errors.ConfigurationError("Incomplete configuration"
" (missing cluster.rsahostkeypub)")
self._config_data = data
- self._config_time = st.st_mtime
- self._config_size = st.st_size
- self._config_inode = st.st_ino
+ # init the last serial as -1 so that the next write will cause
+ # ssconf update
+ self._last_cluster_serial = -1
def _DistributeConfig(self):
"""Distribute the configuration to the other nodes.
if self._offline:
return True
bad = False
- nodelist = self._UnlockedGetNodeList()
- myhostname = self._my_hostname
- try:
- nodelist.remove(myhostname)
- except ValueError:
- pass
+ node_list = []
+ addr_list = []
+ myhostname = self._my_hostname
+ # we can skip checking whether _UnlockedGetNodeInfo returns None
+ # since the node list comes from _UnlocketGetNodeList, and we are
+ # called with the lock held, so no modifications should take place
+ # in between
+ for node_name in self._UnlockedGetNodeList():
+ if node_name == myhostname:
+ continue
+ node_info = self._UnlockedGetNodeInfo(node_name)
+ if not node_info.master_candidate:
+ continue
+ node_list.append(node_info.name)
+ addr_list.append(node_info.primary_ip)
- result = rpc.call_upload_file(nodelist, self._cfg_file)
- for node in nodelist:
+ result = rpc.RpcRunner.call_upload_file(node_list, self._cfg_file,
+ address_list=addr_list)
+ for node in node_list:
if not result[node]:
logging.error("copy of file %s to node %s failed",
self._cfg_file, node)
# we don't need to do os.close(fd) as f.close() did it
os.rename(name, destination)
self.write_count += 1
- # re-set our cache as not to re-read the config file
- try:
- st = os.stat(destination)
- except OSError, err:
- raise errors.ConfigurationError("Can't stat config file: %s" % err)
- self._config_time = st.st_mtime
- self._config_size = st.st_size
- self._config_inode = st.st_ino
- # and redistribute the config file
+
+ # and redistribute the config file to master candidates
self._DistributeConfig()
+ # Write ssconf files on all nodes (including locally)
+ if self._last_cluster_serial < self._config_data.cluster.serial_no:
+ if not self._offline:
+ rpc.RpcRunner.call_write_ssconf_files(self._UnlockedGetNodeList(),
+ self._UnlockedGetSsconfValues())
+ self._last_cluster_serial = self._config_data.cluster.serial_no
+
+ def _UnlockedGetSsconfValues(self):
+ """Return the values needed by ssconf.
+
+ @rtype: dict
+ @return: a dictionary with keys the ssconf names and values their
+ associated value
+
+ """
+ node_list = utils.NiceSort(self._UnlockedGetNodeList())
+ mc_list = [self._UnlockedGetNodeInfo(name) for name in node_list]
+ mc_list = [node.name for node in mc_list if node.master_candidate]
+ node_list = "\n".join(node_list)
+ mc_list = "\n".join(mc_list)
+
+ cluster = self._config_data.cluster
+ return {
+ constants.SS_CLUSTER_NAME: cluster.cluster_name,
+ constants.SS_FILE_STORAGE_DIR: cluster.file_storage_dir,
+ constants.SS_MASTER_CANDIDATES: mc_list,
+ constants.SS_MASTER_IP: cluster.master_ip,
+ constants.SS_MASTER_NETDEV: cluster.master_netdev,
+ constants.SS_MASTER_NODE: cluster.master_node,
+ constants.SS_NODE_LIST: node_list,
+ }
+
@locking.ssynchronized(_config_lock)
- def InitConfig(self, cluster_config, master_node_config):
+ def InitConfig(self, version, cluster_config, master_node_config):
"""Create the initial cluster configuration.
It will contain the current node, which will also be the master
node, and no instances.
+ @type version: int
+ @param version: Configuration version
@type cluster_config: objects.Cluster
@param cluster_config: Cluster configuration
@type master_node_config: objects.Node
master_node_config.name: master_node_config,
}
- self._config_data = objects.ConfigData(cluster=cluster_config,
+ self._config_data = objects.ConfigData(version=version,
+ cluster=cluster_config,
nodes=nodes,
instances={},
serial_no=1)
"""Return the volume group name.
"""
- self._OpenConfig()
return self._config_data.cluster.volume_group_name
@locking.ssynchronized(_config_lock)
"""Set the volume group name.
"""
- self._OpenConfig()
self._config_data.cluster.volume_group_name = vg_name
self._config_data.cluster.serial_no += 1
self._WriteConfig()
"""Return the default bridge.
"""
- self._OpenConfig()
return self._config_data.cluster.default_bridge
@locking.ssynchronized(_config_lock, shared=1)
"""Return the mac prefix.
"""
- self._OpenConfig()
return self._config_data.cluster.mac_prefix
@locking.ssynchronized(_config_lock, shared=1)
the cluster object
"""
- self._OpenConfig()
-
return self._config_data.cluster
@locking.ssynchronized(_config_lock)
if self._config_data is None:
raise errors.ProgrammerError("Configuration file not read,"
" cannot save.")
+ update_serial = False
if isinstance(target, objects.Cluster):
test = target == self._config_data.cluster
elif isinstance(target, objects.Node):
test = target in self._config_data.nodes.values()
+ update_serial = True
elif isinstance(target, objects.Instance):
test = target in self._config_data.instances.values()
else:
" has been read or unknown object")
target.serial_no += 1
+ if update_serial:
+ # for node updates, we need to increase the cluster serial too
+ self._config_data.cluster.serial_no += 1
+
self._WriteConfig()
projects / ganeti-local / blobdiff