X-Git-Url: https://code.grnet.gr/git/ganeti-local/blobdiff_plain/3a93eebbde19c2a11153bb886b5b492b3523a73d..81f7ea25fa942612873eb159e24c44e24ce84e69:/lib/config.py?ds=sidebyside diff --git a/lib/config.py b/lib/config.py index 64ec123..f3eea73 100644 --- a/lib/config.py +++ b/lib/config.py @@ -1,7 +1,7 @@ # # -# Copyright (C) 2006, 2007, 2008, 2009, 2010, 2011 Google Inc. +# Copyright (C) 2006, 2007, 2008, 2009, 2010, 2011, 2012 Google Inc. # # This program is free software; you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by @@ -31,13 +31,14 @@ much memory. """ -# pylint: disable-msg=R0904 +# pylint: disable=R0904 # R0904: Too many public methods import os import random import logging import time +import itertools from ganeti import errors from ganeti import locking @@ -132,6 +133,26 @@ def _MatchNameComponentIgnoreCase(short_name, names): return utils.MatchNameComponent(short_name, names, case_sensitive=False) +def _CheckInstanceDiskIvNames(disks): + """Checks if instance's disks' C{iv_name} attributes are in order. + + @type disks: list of L{objects.Disk} + @param disks: List of disks + @rtype: list of tuples; (int, string, string) + @return: List of wrongly named disks, each tuple contains disk index, + expected and actual name + + """ + result = [] + + for (idx, disk) in enumerate(disks): + exp_iv_name = "disk/%s" % idx + if disk.iv_name != exp_iv_name: + result.append((idx, exp_iv_name, disk.iv_name)) + + return result + + class ConfigWriter: """The interface to the cluster configuration. @@ -164,8 +185,21 @@ class ConfigWriter: self._my_hostname = netutils.Hostname.GetSysName() self._last_cluster_serial = -1 self._cfg_id = None + self._context = None self._OpenConfig(accept_foreign) + def _GetRpc(self, address_list): + """Returns RPC runner for configuration. + + """ + return rpc.ConfigRunner(self._context, address_list) + + def SetContext(self, context): + """Sets Ganeti context. + + """ + self._context = context + # this method needs to be static, so that we can call it on the class @staticmethod def IsCluster(): @@ -189,7 +223,7 @@ class ConfigWriter: def GetNdParams(self, node): """Get the node params populated with cluster defaults. - @type node: L{object.Node} + @type node: L{objects.Node} @param node: The node we want to know the params for @return: A dict with the filled in node params @@ -219,7 +253,7 @@ class ConfigWriter: if mac in all_macs: raise errors.ReservationError("mac already in use") else: - self._temporary_macs.Reserve(mac, ec_id) + self._temporary_macs.Reserve(ec_id, mac) @locking.ssynchronized(_config_lock, shared=1) def ReserveLV(self, lv_name, ec_id): @@ -233,7 +267,7 @@ class ConfigWriter: if lv_name in all_lvs: raise errors.ReservationError("LV already in use") else: - self._temporary_lvs.Reserve(lv_name, ec_id) + self._temporary_lvs.Reserve(ec_id, lv_name) @locking.ssynchronized(_config_lock, shared=1) def GenerateDRBDSecret(self, ec_id): @@ -374,7 +408,7 @@ class ConfigWriter: configuration errors """ - # pylint: disable-msg=R0914 + # pylint: disable=R0914 result = [] seen_macs = [] ports = {} @@ -412,6 +446,28 @@ class ConfigWriter: except errors.ConfigurationError, err: result.append("%s has invalid nicparams: %s" % (owner, err)) + def _helper_ipolicy(owner, params): + try: + objects.InstancePolicy.CheckParameterSyntax(params) + except errors.ConfigurationError, err: + result.append("%s has invalid instance policy: %s" % (owner, err)) + + def _helper_ispecs(owner, params): + for key, value in params.items(): + if key in constants.IPOLICY_ISPECS: + fullkey = "ipolicy/" + key + _helper(owner, fullkey, value, constants.ISPECS_PARAMETER_TYPES) + else: + # FIXME: assuming list type + if key in constants.IPOLICY_PARAMETERS: + exp_type = float + else: + exp_type = list + if not isinstance(value, exp_type): + result.append("%s has invalid instance policy: for %s," + " expecting %s, got %s" % + (owner, key, exp_type.__name__, type(value))) + # check cluster parameters _helper("cluster", "beparams", cluster.SimpleFillBE({}), constants.BES_PARAMETER_TYPES) @@ -420,6 +476,8 @@ class ConfigWriter: _helper_nic("cluster", cluster.SimpleFillNIC({})) _helper("cluster", "ndparams", cluster.SimpleFillND({}), constants.NDS_PARAMETER_TYPES) + _helper_ipolicy("cluster", cluster.SimpleFillIPolicy({})) + _helper_ispecs("cluster", cluster.SimpleFillIPolicy({})) # per-instance checks for instance_name in data.instances: @@ -453,12 +511,12 @@ class ConfigWriter: cluster.FillBE(instance), constants.BES_PARAMETER_TYPES) # gather the drbd ports for duplicate checks - for dsk in instance.disks: + for (idx, dsk) in enumerate(instance.disks): if dsk.dev_type in constants.LDS_DRBD: tcp_port = dsk.logical_id[2] if tcp_port not in ports: ports[tcp_port] = [] - ports[tcp_port].append((instance.name, "drbd disk %s" % dsk.iv_name)) + ports[tcp_port].append((instance.name, "drbd disk %s" % idx)) # gather network port reservation net_port = getattr(instance, "network_port", None) if net_port is not None: @@ -472,6 +530,15 @@ class ConfigWriter: (instance.name, idx, msg) for msg in disk.Verify()]) result.extend(self._CheckDiskIDs(disk, seen_lids, seen_pids)) + wrong_names = _CheckInstanceDiskIvNames(instance.disks) + if wrong_names: + tmp = "; ".join(("name of disk %s should be '%s', but is '%s'" % + (idx, exp_name, actual_name)) + for (idx, exp_name, actual_name) in wrong_names) + + result.append("Instance '%s' has wrongly named disks: %s" % + (instance.name, tmp)) + # cluster-wide pool of free ports for free_port in cluster.tcpudp_port_pool: if free_port not in ports: @@ -534,12 +601,14 @@ class ConfigWriter: result.append("duplicate node group name '%s'" % nodegroup.name) else: nodegroups_names.add(nodegroup.name) + group_name = "group %s" % nodegroup.name + _helper_ipolicy(group_name, cluster.SimpleFillIPolicy(nodegroup.ipolicy)) + _helper_ispecs(group_name, cluster.SimpleFillIPolicy(nodegroup.ipolicy)) if nodegroup.ndparams: - _helper("group %s" % nodegroup.name, "ndparams", + _helper(group_name, "ndparams", cluster.SimpleFillND(nodegroup.ndparams), constants.NDS_PARAMETER_TYPES) - # drbd minors check _, duplicates = self._UnlockedComputeDRBDMap() for node, minor, instance_a, instance_b in duplicates: @@ -878,6 +947,20 @@ class ConfigWriter: return self._config_data.cluster.master_netdev @locking.ssynchronized(_config_lock, shared=1) + def GetMasterNetmask(self): + """Get the netmask of the master node for this cluster. + + """ + return self._config_data.cluster.master_netmask + + @locking.ssynchronized(_config_lock, shared=1) + def GetUseExternalMipScript(self): + """Get flag representing whether to use the external master IP setup script. + + """ + return self._config_data.cluster.use_external_mip_script + + @locking.ssynchronized(_config_lock, shared=1) def GetFileStorageDir(self): """Get the file storage dir for this cluster. @@ -924,6 +1007,23 @@ class ConfigWriter: """ return self._config_data.cluster.primary_ip_family + @locking.ssynchronized(_config_lock, shared=1) + def GetMasterNetworkParameters(self): + """Get network parameters of the master node. + + @rtype: L{object.MasterNetworkParameters} + @return: network parameters of the master node + + """ + cluster = self._config_data.cluster + result = objects.MasterNetworkParameters(name=cluster.master_node, + ip=cluster.master_ip, + netmask=cluster.master_netmask, + netdev=cluster.master_netdev, + ip_family=cluster.primary_ip_family) + + return result + @locking.ssynchronized(_config_lock) def AddNodeGroup(self, group, ec_id, check_uuid=True): """Add a node group to the configuration. @@ -1071,6 +1171,17 @@ class ConfigWriter: """ return self._config_data.nodegroups.keys() + @locking.ssynchronized(_config_lock, shared=1) + def GetNodeGroupMembersByNodes(self, nodes): + """Get nodes which are member in the same nodegroups as the given nodes. + + """ + ngfn = lambda node_name: self._UnlockedGetNodeInfo(node_name).group + return frozenset(member_name + for node_name in nodes + for member_name in + self._UnlockedGetNodeGroup(ngfn(node_name)).members) + @locking.ssynchronized(_config_lock) def AddInstance(self, instance, ec_id): """Add an instance to the config. @@ -1121,15 +1232,15 @@ class ConfigWriter: """Set the instance's status to a given value. """ - assert isinstance(status, bool), \ + assert status in constants.ADMINST_ALL, \ "Invalid status '%s' passed to SetInstanceStatus" % (status,) if instance_name not in self._config_data.instances: raise errors.ConfigurationError("Unknown instance '%s'" % instance_name) instance = self._config_data.instances[instance_name] - if instance.admin_up != status: - instance.admin_up = status + if instance.admin_state != status: + instance.admin_state = status instance.serial_no += 1 instance.mtime = time.time() self._WriteConfig() @@ -1139,7 +1250,14 @@ class ConfigWriter: """Mark the instance status to up in the config. """ - self._SetInstanceStatus(instance_name, True) + self._SetInstanceStatus(instance_name, constants.ADMINST_UP) + + @locking.ssynchronized(_config_lock) + def MarkInstanceOffline(self, instance_name): + """Mark the instance status to down in the config. + + """ + self._SetInstanceStatus(instance_name, constants.ADMINST_OFFLINE) @locking.ssynchronized(_config_lock) def RemoveInstance(self, instance_name): @@ -1148,6 +1266,14 @@ class ConfigWriter: """ if instance_name not in self._config_data.instances: raise errors.ConfigurationError("Unknown instance '%s'" % instance_name) + + # If a network port has been allocated to the instance, + # return it to the pool of free ports. + inst = self._config_data.instances[instance_name] + network_port = getattr(inst, "network_port", None) + if network_port is not None: + self._config_data.cluster.tcpudp_port_pool.add(network_port) + del self._config_data.instances[instance_name] self._config_data.cluster.serial_no += 1 self._WriteConfig() @@ -1163,24 +1289,27 @@ class ConfigWriter: """ if old_name not in self._config_data.instances: raise errors.ConfigurationError("Unknown instance '%s'" % old_name) - inst = self._config_data.instances[old_name] - del self._config_data.instances[old_name] + + # Operate on a copy to not loose instance object in case of a failure + inst = self._config_data.instances[old_name].Copy() inst.name = new_name - for disk in inst.disks: + for (idx, disk) in enumerate(inst.disks): if disk.dev_type == constants.LD_FILE: # rename the file paths in logical and physical id file_storage_dir = os.path.dirname(os.path.dirname(disk.logical_id[1])) - disk_fname = "disk%s" % disk.iv_name.split("/")[1] - disk.physical_id = disk.logical_id = (disk.logical_id[0], - utils.PathJoin(file_storage_dir, - inst.name, - disk_fname)) + disk.logical_id = (disk.logical_id[0], + utils.PathJoin(file_storage_dir, inst.name, + "disk%s" % idx)) + disk.physical_id = disk.logical_id + + # Actually replace instance object + del self._config_data.instances[old_name] + self._config_data.instances[inst.name] = inst # Force update of ssconf files self._config_data.cluster.serial_no += 1 - self._config_data.instances[inst.name] = inst self._WriteConfig() @locking.ssynchronized(_config_lock) @@ -1188,7 +1317,7 @@ class ConfigWriter: """Mark the status of an instance to down in the configuration. """ - self._SetInstanceStatus(instance_name, False) + self._SetInstanceStatus(instance_name, constants.ADMINST_DOWN) def _UnlockedGetInstanceList(self): """Get the list of instances. @@ -1243,6 +1372,38 @@ class ConfigWriter: return self._UnlockedGetInstanceInfo(instance_name) @locking.ssynchronized(_config_lock, shared=1) + def GetInstanceNodeGroups(self, instance_name, primary_only=False): + """Returns set of node group UUIDs for instance's nodes. + + @rtype: frozenset + + """ + instance = self._UnlockedGetInstanceInfo(instance_name) + if not instance: + raise errors.ConfigurationError("Unknown instance '%s'" % instance_name) + + if primary_only: + nodes = [instance.primary_node] + else: + nodes = instance.all_nodes + + return frozenset(self._UnlockedGetNodeInfo(node_name).group + for node_name in nodes) + + @locking.ssynchronized(_config_lock, shared=1) + def GetMultiInstanceInfo(self, instances): + """Get the configuration of multiple instances. + + @param instances: list of instance names + @rtype: list + @return: list of tuples (instance, instance_info), where + instance_info is what would GetInstanceInfo return for the + node, while keeping the original order + + """ + return [(name, self._UnlockedGetInstanceInfo(name)) for name in instances] + + @locking.ssynchronized(_config_lock, shared=1) def GetAllInstancesInfo(self): """Get the configuration of all instances. @@ -1255,6 +1416,22 @@ class ConfigWriter: for instance in self._UnlockedGetInstanceList()]) return my_dict + @locking.ssynchronized(_config_lock, shared=1) + def GetInstancesInfoByFilter(self, filter_fn): + """Get instance configuration with a filter. + + @type filter_fn: callable + @param filter_fn: Filter function receiving instance object as parameter, + returning boolean. Important: this function is called while the + configuration locks is held. It must not do any complex work or call + functions potentially leading to a deadlock. Ideally it doesn't call any + other functions and just compares instance attributes. + + """ + return dict((name, inst) + for (name, inst) in self._config_data.instances.items() + if filter_fn(inst)) + @locking.ssynchronized(_config_lock) def AddNode(self, node, ec_id): """Add a node to the configuration. @@ -1346,6 +1523,26 @@ class ConfigWriter: sec.append(inst.name) return (pri, sec) + @locking.ssynchronized(_config_lock, shared=1) + def GetNodeGroupInstances(self, uuid, primary_only=False): + """Get the instances of a node group. + + @param uuid: Node group UUID + @param primary_only: Whether to only consider primary nodes + @rtype: frozenset + @return: List of instance names in node group + + """ + if primary_only: + nodes_fn = lambda inst: [inst.primary_node] + else: + nodes_fn = lambda inst: inst.all_nodes + + return frozenset(inst.name + for inst in self._config_data.instances.values() + for node_name in nodes_fn(inst) + if self._UnlockedGetNodeInfo(node_name).group == uuid) + def _UnlockedGetNodeList(self): """Return the list of nodes which are in the configuration. @@ -1398,6 +1595,19 @@ class ConfigWriter: return [node.name for node in all_nodes if not node.vm_capable] @locking.ssynchronized(_config_lock, shared=1) + def GetMultiNodeInfo(self, nodes): + """Get the configuration of multiple nodes. + + @param nodes: list of node names + @rtype: list + @return: list of tuples of (node, node_info), where node_info is + what would GetNodeInfo return for the node, in the original + order + + """ + return [(name, self._UnlockedGetNodeInfo(name)) for name in nodes] + + @locking.ssynchronized(_config_lock, shared=1) def GetAllNodesInfo(self): """Get the configuration of all nodes. @@ -1406,9 +1616,27 @@ class ConfigWriter: would GetNodeInfo return for the node """ - my_dict = dict([(node, self._UnlockedGetNodeInfo(node)) - for node in self._UnlockedGetNodeList()]) - return my_dict + return self._UnlockedGetAllNodesInfo() + + def _UnlockedGetAllNodesInfo(self): + """Gets configuration of all nodes. + + @note: See L{GetAllNodesInfo} + + """ + return dict([(node, self._UnlockedGetNodeInfo(node)) + for node in self._UnlockedGetNodeList()]) + + @locking.ssynchronized(_config_lock, shared=1) + def GetNodeGroupsFromNodes(self, nodes): + """Returns groups for a list of nodes. + + @type nodes: list of string + @param nodes: List of node names + @rtype: frozenset + + """ + return frozenset(self._UnlockedGetNodeInfo(name).group for name in nodes) def _UnlockedGetMasterCandidateStats(self, exceptions=None): """Get the number of current and maximum desired and possible candidates. @@ -1508,6 +1736,79 @@ class ConfigWriter: else: nodegroup_obj.members.remove(node.name) + @locking.ssynchronized(_config_lock) + def AssignGroupNodes(self, mods): + """Changes the group of a number of nodes. + + @type mods: list of tuples; (node name, new group UUID) + @param mods: Node membership modifications + + """ + groups = self._config_data.nodegroups + nodes = self._config_data.nodes + + resmod = [] + + # Try to resolve names/UUIDs first + for (node_name, new_group_uuid) in mods: + try: + node = nodes[node_name] + except KeyError: + raise errors.ConfigurationError("Unable to find node '%s'" % node_name) + + if node.group == new_group_uuid: + # Node is being assigned to its current group + logging.debug("Node '%s' was assigned to its current group (%s)", + node_name, node.group) + continue + + # Try to find current group of node + try: + old_group = groups[node.group] + except KeyError: + raise errors.ConfigurationError("Unable to find old group '%s'" % + node.group) + + # Try to find new group for node + try: + new_group = groups[new_group_uuid] + except KeyError: + raise errors.ConfigurationError("Unable to find new group '%s'" % + new_group_uuid) + + assert node.name in old_group.members, \ + ("Inconsistent configuration: node '%s' not listed in members for its" + " old group '%s'" % (node.name, old_group.uuid)) + assert node.name not in new_group.members, \ + ("Inconsistent configuration: node '%s' already listed in members for" + " its new group '%s'" % (node.name, new_group.uuid)) + + resmod.append((node, old_group, new_group)) + + # Apply changes + for (node, old_group, new_group) in resmod: + assert node.uuid != new_group.uuid and old_group.uuid != new_group.uuid, \ + "Assigning to current group is not possible" + + node.group = new_group.uuid + + # Update members of involved groups + if node.name in old_group.members: + old_group.members.remove(node.name) + if node.name not in new_group.members: + new_group.members.append(node.name) + + # Update timestamps and serials (only once per node/group object) + now = time.time() + for obj in frozenset(itertools.chain(*resmod)): # pylint: disable=W0142 + obj.serial_no += 1 + obj.mtime = now + + # Force ssconf update + self._config_data.cluster.serial_no += 1 + + self._WriteConfig() + def _BumpSerialNo(self): """Bump up the serial number of the config. @@ -1538,8 +1839,8 @@ class ConfigWriter: # Make sure the configuration has the right version _ValidateConfig(data) - if (not hasattr(data, 'cluster') or - not hasattr(data.cluster, 'rsahostkeypub')): + if (not hasattr(data, "cluster") or + not hasattr(data.cluster, "rsahostkeypub")): raise errors.ConfigurationError("Incomplete configuration" " (missing cluster.rsahostkeypub)") @@ -1630,8 +1931,9 @@ class ConfigWriter: node_list.append(node_info.name) addr_list.append(node_info.primary_ip) - result = rpc.RpcRunner.call_upload_file(node_list, self._cfg_file, - address_list=addr_list) + # TODO: Use dedicated resolver talking to config writer for name resolution + result = \ + self._GetRpc(addr_list).call_upload_file(node_list, self._cfg_file) for to_node, to_result in result.items(): msg = to_result.fail_msg if msg: @@ -1690,7 +1992,7 @@ class ConfigWriter: # Write ssconf files on all nodes (including locally) if self._last_cluster_serial < self._config_data.cluster.serial_no: if not self._offline: - result = rpc.RpcRunner.call_write_ssconf_files( + result = self._GetRpc(None).call_write_ssconf_files( self._UnlockedGetOnlineNodeList(), self._UnlockedGetSsconfValues()) @@ -1744,7 +2046,7 @@ class ConfigWriter: self._config_data.nodegroups.values()] nodegroups_data = fn(utils.NiceSort(nodegroups)) - return { + ssconf_values = { constants.SS_CLUSTER_NAME: cluster.cluster_name, constants.SS_CLUSTER_TAGS: cluster_tags, constants.SS_FILE_STORAGE_DIR: cluster.file_storage_dir, @@ -1753,6 +2055,7 @@ class ConfigWriter: constants.SS_MASTER_CANDIDATES_IPS: mc_ips_data, constants.SS_MASTER_IP: cluster.master_ip, constants.SS_MASTER_NETDEV: cluster.master_netdev, + constants.SS_MASTER_NETMASK: str(cluster.master_netmask), constants.SS_MASTER_NODE: cluster.master_node, constants.SS_NODE_LIST: node_data, constants.SS_NODE_PRIMARY_IPS: node_pri_ips_data, @@ -1767,6 +2070,13 @@ class ConfigWriter: constants.SS_UID_POOL: uid_pool, constants.SS_NODEGROUPS: nodegroups_data, } + bad_values = [(k, v) for k, v in ssconf_values.items() + if not isinstance(v, (str, basestring))] + if bad_values: + err = utils.CommaJoin("%s=%s" % (k, v) for k, v in bad_values) + raise errors.ConfigurationError("Some ssconf key(s) have non-string" + " values: %s" % err) + return ssconf_values @locking.ssynchronized(_config_lock, shared=1) def GetSsconfValues(self):