X-Git-Url: https://code.grnet.gr/git/ganeti-local/blobdiff_plain/90e99856b5f0f833fe016ef9203f46329a11491e..b3c728dc6f916757627f3da6422a43318dfecf8b:/lib/config.py diff --git a/lib/config.py b/lib/config.py index 6116b8d..9d87c2b 100644 --- a/lib/config.py +++ b/lib/config.py @@ -1,7 +1,7 @@ # # -# Copyright (C) 2006, 2007, 2008, 2009, 2010 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 @@ -49,6 +50,8 @@ from ganeti import serializer from ganeti import uidpool from ganeti import netutils from ganeti import runtime +from ganeti import pathutils +from ganeti import network _config_lock = locking.SharedLock("ConfigWriter") @@ -105,6 +108,17 @@ class TemporaryReservationManager: all_reserved.update(holder_reserved) return all_reserved + def GetECReserved(self, ec_id): + """ Used when you want to retrieve all reservations for a specific + execution context. E.g when commiting reserved IPs for a specific + network. + + """ + ec_reserved = set() + if ec_id in self._ec_reserved: + ec_reserved.update(self._ec_reserved[ec_id]) + return ec_reserved + def Generate(self, existing, generate_one_fn, ec_id): """Generate a new resource of this type @@ -125,6 +139,33 @@ class TemporaryReservationManager: return new_resource +def _MatchNameComponentIgnoreCase(short_name, names): + """Wrapper around L{utils.text.MatchNameComponent}. + + """ + 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. @@ -139,7 +180,7 @@ class ConfigWriter: self._config_data = None self._offline = offline if cfg_file is None: - self._cfg_file = constants.CLUSTER_CONF_FILE + self._cfg_file = pathutils.CLUSTER_CONF_FILE else: self._cfg_file = cfg_file self._getents = _getents @@ -148,8 +189,10 @@ class ConfigWriter: self._temporary_macs = TemporaryReservationManager() self._temporary_secrets = TemporaryReservationManager() self._temporary_lvs = TemporaryReservationManager() + self._temporary_ips = TemporaryReservationManager() self._all_rms = [self._temporary_ids, self._temporary_macs, - self._temporary_secrets, self._temporary_lvs] + self._temporary_secrets, self._temporary_lvs, + self._temporary_ips] # Note: in order to prevent errors when resolving our name in # _DistributeConfig, we compute it here once and reuse it; it's # better to raise an error before starting to modify the config @@ -157,32 +200,34 @@ 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(): """Check if the cluster is configured. """ - return os.path.exists(constants.CLUSTER_CONF_FILE) - - def _GenerateOneMAC(self): - """Generate one mac address - - """ - prefix = self._config_data.cluster.mac_prefix - byte1 = random.randrange(0, 256) - byte2 = random.randrange(0, 256) - byte3 = random.randrange(0, 256) - mac = "%s:%02x:%02x:%02x" % (prefix, byte1, byte2, byte3) - return mac + return os.path.exists(pathutils.CLUSTER_CONF_FILE) @locking.ssynchronized(_config_lock, shared=1) 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 @@ -191,14 +236,82 @@ class ConfigWriter: return self._config_data.cluster.FillND(node, nodegroup) @locking.ssynchronized(_config_lock, shared=1) - def GenerateMAC(self, ec_id): + def GetInstanceDiskParams(self, instance): + """Get the disk params populated with inherit chain. + + @type instance: L{objects.Instance} + @param instance: The instance we want to know the params for + @return: A dict with the filled in disk params + + """ + node = self._UnlockedGetNodeInfo(instance.primary_node) + nodegroup = self._UnlockedGetNodeGroup(node.group) + return self._UnlockedGetGroupDiskParams(nodegroup) + + @locking.ssynchronized(_config_lock, shared=1) + def GetGroupDiskParams(self, group): + """Get the disk params populated with inherit chain. + + @type group: L{objects.NodeGroup} + @param group: The group we want to know the params for + @return: A dict with the filled in disk params + + """ + return self._UnlockedGetGroupDiskParams(group) + + def _UnlockedGetGroupDiskParams(self, group): + """Get the disk params populated with inherit chain down to node-group. + + @type group: L{objects.NodeGroup} + @param group: The group we want to know the params for + @return: A dict with the filled in disk params + + """ + return self._config_data.cluster.SimpleFillDP(group.diskparams) + + def _UnlockedGetNetworkMACPrefix(self, net): + """Return the network mac prefix if it exists or the cluster level default. + + """ + prefix = None + if net: + net_uuid = self._UnlockedLookupNetwork(net) + if net_uuid: + nobj = self._UnlockedGetNetwork(net_uuid) + if nobj.mac_prefix: + prefix = nobj.mac_prefix + + return prefix + + def _GenerateOneMAC(self, prefix=None): + """Return a function that randomly generates a MAC suffic + and appends it to the given prefix. If prefix is not given get + the cluster level default. + + """ + if not prefix: + prefix = self._config_data.cluster.mac_prefix + + def GenMac(): + byte1 = random.randrange(0, 256) + byte2 = random.randrange(0, 256) + byte3 = random.randrange(0, 256) + mac = "%s:%02x:%02x:%02x" % (prefix, byte1, byte2, byte3) + return mac + + return GenMac + + @locking.ssynchronized(_config_lock, shared=1) + def GenerateMAC(self, net, ec_id): """Generate a MAC for an instance. This should check the current instances for duplicates. """ existing = self._AllMACs() - return self._temporary_ids.Generate(existing, self._GenerateOneMAC, ec_id) + prefix = self._UnlockedGetNetworkMACPrefix(net) + gen_mac = self._GenerateOneMAC(prefix) + return self._temporary_ids.Generate(existing, gen_mac, ec_id) @locking.ssynchronized(_config_lock, shared=1) def ReserveMAC(self, mac, ec_id): @@ -212,7 +325,93 @@ 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) + + def _UnlockedCommitTemporaryIps(self, ec_id): + """Commit all reserved IP address to their respective pools + + """ + for action, address, net_uuid in self._temporary_ips.GetECReserved(ec_id): + self._UnlockedCommitIp(action, net_uuid, address) + + def _UnlockedCommitIp(self, action, net_uuid, address): + """Commit a reserved IP address to an IP pool. + + The IP address is taken from the network's IP pool and marked as reserved. + + """ + nobj = self._UnlockedGetNetwork(net_uuid) + pool = network.AddressPool(nobj) + if action == constants.RESERVE_ACTION: + pool.Reserve(address) + elif action == constants.RELEASE_ACTION: + pool.Release(address) + + def _UnlockedReleaseIp(self, net_uuid, address, ec_id): + """Give a specific IP address back to an IP pool. + + The IP address is returned to the IP pool designated by pool_id and marked + as reserved. + + """ + self._temporary_ips.Reserve(ec_id, + (constants.RELEASE_ACTION, address, net_uuid)) + + @locking.ssynchronized(_config_lock, shared=1) + def ReleaseIp(self, net, address, ec_id): + """Give a specified IP address back to an IP pool. + + This is just a wrapper around _UnlockedReleaseIp. + + """ + net_uuid = self._UnlockedLookupNetwork(net) + if net_uuid: + self._UnlockedReleaseIp(net_uuid, address, ec_id) + + @locking.ssynchronized(_config_lock, shared=1) + def GenerateIp(self, net, ec_id): + """Find a free IPv4 address for an instance. + + """ + net_uuid = self._UnlockedLookupNetwork(net) + nobj = self._UnlockedGetNetwork(net_uuid) + pool = network.AddressPool(nobj) + + def gen_one(): + try: + ip = pool.GenerateFree() + except errors.AddressPoolError: + raise errors.ReservationError("Cannot generate IP. Network is full") + return (constants.RESERVE_ACTION, ip, net_uuid) + + _, address, _ = self._temporary_ips.Generate([], gen_one, ec_id) + return address + + def _UnlockedReserveIp(self, net_uuid, address, ec_id): + """Reserve a given IPv4 address for use by an instance. + + """ + nobj = self._UnlockedGetNetwork(net_uuid) + pool = network.AddressPool(nobj) + try: + isreserved = pool.IsReserved(address) + except errors.AddressPoolError: + raise errors.ReservationError("IP address not in network") + if isreserved: + raise errors.ReservationError("IP address already in use") + + return self._temporary_ips.Reserve(ec_id, + (constants.RESERVE_ACTION, + address, net_uuid)) + + @locking.ssynchronized(_config_lock, shared=1) + def ReserveIp(self, net, address, ec_id): + """Reserve a given IPv4 address for use by an instance. + + """ + net_uuid = self._UnlockedLookupNetwork(net) + if net_uuid: + return self._UnlockedReserveIp(net_uuid, address, ec_id) @locking.ssynchronized(_config_lock, shared=1) def ReserveLV(self, lv_name, ec_id): @@ -226,7 +425,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): @@ -367,29 +566,76 @@ class ConfigWriter: configuration errors """ + # pylint: disable=R0914 result = [] seen_macs = [] ports = {} data = self._config_data + cluster = data.cluster seen_lids = [] seen_pids = [] # global cluster checks - if not data.cluster.enabled_hypervisors: + if not cluster.enabled_hypervisors: result.append("enabled hypervisors list doesn't have any entries") - invalid_hvs = set(data.cluster.enabled_hypervisors) - constants.HYPER_TYPES + invalid_hvs = set(cluster.enabled_hypervisors) - constants.HYPER_TYPES if invalid_hvs: result.append("enabled hypervisors contains invalid entries: %s" % invalid_hvs) - missing_hvp = (set(data.cluster.enabled_hypervisors) - - set(data.cluster.hvparams.keys())) + missing_hvp = (set(cluster.enabled_hypervisors) - + set(cluster.hvparams.keys())) if missing_hvp: result.append("hypervisor parameters missing for the enabled" " hypervisor(s) %s" % utils.CommaJoin(missing_hvp)) - if data.cluster.master_node not in data.nodes: + if cluster.master_node not in data.nodes: result.append("cluster has invalid primary node '%s'" % - data.cluster.master_node) + cluster.master_node) + + def _helper(owner, attr, value, template): + try: + utils.ForceDictType(value, template) + except errors.GenericError, err: + result.append("%s has invalid %s: %s" % (owner, attr, err)) + + def _helper_nic(owner, params): + try: + objects.NIC.CheckParameterSyntax(params) + except errors.ConfigurationError, err: + result.append("%s has invalid nicparams: %s" % (owner, err)) + + def _helper_ipolicy(owner, params, check_std): + try: + objects.InstancePolicy.CheckParameterSyntax(params, check_std) + 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) + _helper("cluster", "nicparams", cluster.SimpleFillNIC({}), + constants.NICS_PARAMETER_TYPES) + _helper_nic("cluster", cluster.SimpleFillNIC({})) + _helper("cluster", "ndparams", cluster.SimpleFillND({}), + constants.NDS_PARAMETER_TYPES) + _helper_ipolicy("cluster", cluster.SimpleFillIPolicy({}), True) + _helper_ispecs("cluster", cluster.SimpleFillIPolicy({})) # per-instance checks for instance_name in data.instances: @@ -410,14 +656,25 @@ class ConfigWriter: (instance_name, idx, nic.mac)) else: seen_macs.append(nic.mac) + if nic.nicparams: + filled = cluster.SimpleFillNIC(nic.nicparams) + owner = "instance %s nic %d" % (instance.name, idx) + _helper(owner, "nicparams", + filled, constants.NICS_PARAMETER_TYPES) + _helper_nic(owner, filled) + + # parameter checks + if instance.beparams: + _helper("instance %s" % instance.name, "beparams", + 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: @@ -431,8 +688,17 @@ 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 data.cluster.tcpudp_port_pool: + for free_port in cluster.tcpudp_port_pool: if free_port not in ports: ports[free_port] = [] ports[free_port].append(("cluster", "port marked as free")) @@ -448,11 +714,11 @@ class ConfigWriter: # highest used tcp port check if keys: - if keys[-1] > data.cluster.highest_used_port: + if keys[-1] > cluster.highest_used_port: result.append("Highest used port mismatch, saved %s, computed %s" % - (data.cluster.highest_used_port, keys[-1])) + (cluster.highest_used_port, keys[-1])) - if not data.nodes[data.cluster.master_node].master_candidate: + if not data.nodes[cluster.master_node].master_candidate: result.append("Master node is not a master candidate") # master candidate checks @@ -471,6 +737,13 @@ class ConfigWriter: " drain=%s, offline=%s" % (node.name, node.master_candidate, node.drained, node.offline)) + if node.group not in data.nodegroups: + result.append("Node '%s' has invalid group '%s'" % + (node.name, node.group)) + else: + _helper("node %s" % node.name, "ndparams", + cluster.FillND(node, data.nodegroups[node.group]), + constants.NDS_PARAMETER_TYPES) # nodegroups checks nodegroups_names = set() @@ -486,6 +759,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), + False) + _helper_ispecs(group_name, cluster.SimpleFillIPolicy(nodegroup.ipolicy)) + if nodegroup.ndparams: + _helper(group_name, "ndparams", + cluster.SimpleFillND(nodegroup.ndparams), + constants.NDS_PARAMETER_TYPES) # drbd minors check _, duplicates = self._UnlockedComputeDRBDMap() @@ -494,13 +775,13 @@ class ConfigWriter: " %s and %s" % (minor, node, instance_a, instance_b)) # IP checks - default_nicparams = data.cluster.nicparams[constants.PP_DEFAULT] + default_nicparams = cluster.nicparams[constants.PP_DEFAULT] ips = {} def _AddIpAddress(ip, name): ips.setdefault(ip, []).append(name) - _AddIpAddress(data.cluster.master_ip, "cluster_ip") + _AddIpAddress(cluster.master_ip, "cluster_ip") for node in data.nodes.values(): _AddIpAddress(node.primary_ip, "node:%s/primary" % node.name) @@ -523,7 +804,7 @@ class ConfigWriter: else: raise errors.ProgrammerError("NIC mode '%s' not handled" % nic_mode) - _AddIpAddress("%s/%s" % (link, nic.ip), + _AddIpAddress("%s/%s/%s" % (link, nic.ip, nic.network), "instance:%s/nic:%d" % (instance.name, idx)) for ip, owners in ips.items(): @@ -601,12 +882,15 @@ class ConfigWriter: def AddTcpUdpPort(self, port): """Adds a new port to the available port pool. + @warning: this method does not "flush" the configuration (via + L{_WriteConfig}); callers should do that themselves once the + configuration is stable + """ if not isinstance(port, int): raise errors.ProgrammerError("Invalid type passed for port") self._config_data.cluster.tcpudp_port_pool.add(port) - self._WriteConfig() @locking.ssynchronized(_config_lock, shared=1) def GetPortList(self): @@ -825,6 +1109,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. @@ -832,6 +1130,13 @@ class ConfigWriter: return self._config_data.cluster.file_storage_dir @locking.ssynchronized(_config_lock, shared=1) + def GetSharedFileStorageDir(self): + """Get the shared file storage dir for this cluster. + + """ + return self._config_data.cluster.shared_file_storage_dir + + @locking.ssynchronized(_config_lock, shared=1) def GetHypervisorType(self): """Get the hypervisor type for this cluster. @@ -864,6 +1169,22 @@ 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. @@ -896,6 +1217,16 @@ class ConfigWriter: if check_uuid: self._EnsureUUID(group, ec_id) + try: + existing_uuid = self._UnlockedLookupNodeGroup(group.name) + except errors.OpPrereqError: + pass + else: + raise errors.OpPrereqError("Desired group name '%s' already exists as a" + " node group (UUID: %s)" % + (group.name, existing_uuid), + errors.ECODE_EXISTS) + group.serial_no = 1 group.ctime = group.mtime = time.time() group.UpgradeConfig() @@ -916,12 +1247,14 @@ class ConfigWriter: if group_uuid not in self._config_data.nodegroups: raise errors.ConfigurationError("Unknown node group '%s'" % group_uuid) + assert len(self._config_data.nodegroups) != 1, \ + "Group '%s' is the only group, cannot be removed" % group_uuid + del self._config_data.nodegroups[group_uuid] self._config_data.cluster.serial_no += 1 self._WriteConfig() - @locking.ssynchronized(_config_lock, shared=1) - def LookupNodeGroup(self, target): + def _UnlockedLookupNodeGroup(self, target): """Lookup a node group's UUID. @type target: string or None @@ -934,7 +1267,7 @@ class ConfigWriter: if target is None: if len(self._config_data.nodegroups) != 1: raise errors.OpPrereqError("More than one node group exists. Target" - " group must be specified explicitely.") + " group must be specified explicitly.") else: return self._config_data.nodegroups.keys()[0] if target in self._config_data.nodegroups: @@ -945,6 +1278,20 @@ class ConfigWriter: raise errors.OpPrereqError("Node group '%s' not found" % target, errors.ECODE_NOENT) + @locking.ssynchronized(_config_lock, shared=1) + def LookupNodeGroup(self, target): + """Lookup a node group's UUID. + + This function is just a wrapper over L{_UnlockedLookupNodeGroup}. + + @type target: string or None + @param target: group name or UUID or None to look for the default + @rtype: string + @return: nodegroup UUID + + """ + return self._UnlockedLookupNodeGroup(target) + def _UnlockedGetNodeGroup(self, uuid): """Lookup a node group. @@ -985,6 +1332,28 @@ 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, shared=1) + def GetMultiNodeGroupInfo(self, group_uuids): + """Get the configuration of multiple node groups. + + @param group_uuids: List of node group UUIDs + @rtype: list + @return: List of tuples of (group_uuid, group_info) + + """ + return [(uuid, self._UnlockedGetNodeGroup(uuid)) for uuid in group_uuids] + @locking.ssynchronized(_config_lock) def AddInstance(self, instance, ec_id): """Add an instance to the config. @@ -1016,6 +1385,7 @@ class ConfigWriter: self._config_data.instances[instance.name] = instance self._config_data.cluster.serial_no += 1 self._UnlockedReleaseDRBDMinors(instance.name) + self._UnlockedCommitTemporaryIps(ec_id) self._WriteConfig() def _EnsureUUID(self, item, ec_id): @@ -1035,15 +1405,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() @@ -1053,7 +1423,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): @@ -1062,6 +1439,23 @@ 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) + + instance = self._UnlockedGetInstanceInfo(instance_name) + + for nic in instance.nics: + if nic.network is not None and nic.ip is not None: + net_uuid = self._UnlockedLookupNetwork(nic.network) + if net_uuid: + # Return all IP addresses to the respective address pools + self._UnlockedCommitIp(constants.RELEASE_ACTION, net_uuid, nic.ip) + del self._config_data.instances[instance_name] self._config_data.cluster.serial_no += 1 self._WriteConfig() @@ -1077,24 +1471,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) @@ -1102,7 +1499,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. @@ -1122,14 +1519,12 @@ class ConfigWriter: """ return self._UnlockedGetInstanceList() - @locking.ssynchronized(_config_lock, shared=1) def ExpandInstanceName(self, short_name): """Attempt to expand an incomplete instance name. """ - return utils.MatchNameComponent(short_name, - self._config_data.instances.keys(), - case_sensitive=False) + # Locking is done in L{ConfigWriter.GetInstanceList} + return _MatchNameComponentIgnoreCase(short_name, self.GetInstanceList()) def _UnlockedGetInstanceInfo(self, instance_name): """Returns information about an instance. @@ -1159,6 +1554,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. @@ -1171,6 +1598,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. @@ -1205,14 +1648,12 @@ class ConfigWriter: self._config_data.cluster.serial_no += 1 self._WriteConfig() - @locking.ssynchronized(_config_lock, shared=1) def ExpandNodeName(self, short_name): - """Attempt to expand an incomplete instance name. + """Attempt to expand an incomplete node name. """ - return utils.MatchNameComponent(short_name, - self._config_data.nodes.keys(), - case_sensitive=False) + # Locking is done in L{ConfigWriter.GetNodeList} + return _MatchNameComponentIgnoreCase(short_name, self.GetNodeList()) def _UnlockedGetNodeInfo(self, node_name): """Get the configuration of a node, as stored in the config. @@ -1264,6 +1705,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. @@ -1298,6 +1759,15 @@ class ConfigWriter: return self._UnlockedGetOnlineNodeList() @locking.ssynchronized(_config_lock, shared=1) + def GetVmCapableNodeList(self): + """Return the list of nodes which are not vm capable. + + """ + all_nodes = [self._UnlockedGetNodeInfo(node) + for node in self._UnlockedGetNodeList()] + return [node.name for node in all_nodes if node.vm_capable] + + @locking.ssynchronized(_config_lock, shared=1) def GetNonVmCapableNodeList(self): """Return the list of nodes which are not vm capable. @@ -1307,6 +1777,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. @@ -1315,9 +1798,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. @@ -1417,6 +1918,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. @@ -1447,8 +2021,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)") @@ -1539,8 +2113,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: @@ -1599,7 +2174,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()) @@ -1652,15 +2227,20 @@ class ConfigWriter: nodegroups = ["%s %s" % (nodegroup.uuid, nodegroup.name) for nodegroup in self._config_data.nodegroups.values()] nodegroups_data = fn(utils.NiceSort(nodegroups)) + networks = ["%s %s" % (net.uuid, net.name) for net in + self._config_data.networks.values()] + networks_data = fn(utils.NiceSort(networks)) - 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, + constants.SS_SHARED_FILE_STORAGE_DIR: cluster.shared_file_storage_dir, constants.SS_MASTER_CANDIDATES: mc_data, 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, @@ -1674,7 +2254,15 @@ class ConfigWriter: constants.SS_MAINTAIN_NODE_HEALTH: str(cluster.maintain_node_health), constants.SS_UID_POOL: uid_pool, constants.SS_NODEGROUPS: nodegroups_data, + constants.SS_NETWORKS: networks_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): @@ -1740,7 +2328,7 @@ class ConfigWriter: return self._config_data.HasAnyDiskOfType(dev_type) @locking.ssynchronized(_config_lock) - def Update(self, target, feedback_fn): + def Update(self, target, feedback_fn, ec_id=None): """Notify function to be called after updates. This function must be called when an object (as returned by @@ -1768,6 +2356,8 @@ class ConfigWriter: test = target in self._config_data.instances.values() elif isinstance(target, objects.NodeGroup): test = target in self._config_data.nodegroups.values() + elif isinstance(target, objects.Network): + test = target in self._config_data.networks.values() else: raise errors.ProgrammerError("Invalid object type (%s) passed to" " ConfigWriter.Update" % type(target)) @@ -1785,6 +2375,10 @@ class ConfigWriter: if isinstance(target, objects.Instance): self._UnlockedReleaseDRBDMinors(target.name) + if ec_id is not None: + # Commit all ips reserved by OpInstanceSetParams and OpGroupSetParams + self._UnlockedCommitTemporaryIps(ec_id) + self._WriteConfig(feedback_fn=feedback_fn) @locking.ssynchronized(_config_lock) @@ -1794,3 +2388,199 @@ class ConfigWriter: """ for rm in self._all_rms: rm.DropECReservations(ec_id) + + @locking.ssynchronized(_config_lock, shared=1) + def GetAllNetworksInfo(self): + """Get configuration info of all the networks. + + """ + return dict(self._config_data.networks) + + def _UnlockedGetNetworkList(self): + """Get the list of networks. + + This function is for internal use, when the config lock is already held. + + """ + return self._config_data.networks.keys() + + @locking.ssynchronized(_config_lock, shared=1) + def GetNetworkList(self): + """Get the list of networks. + + @return: array of networks, ex. ["main", "vlan100", "200] + + """ + return self._UnlockedGetNetworkList() + + @locking.ssynchronized(_config_lock, shared=1) + def GetNetworkNames(self): + """Get a list of network names + + """ + names = [net.name + for net in self._config_data.networks.values()] + return names + + def _UnlockedGetNetwork(self, uuid): + """Returns information about a network. + + This function is for internal use, when the config lock is already held. + + """ + if uuid not in self._config_data.networks: + return None + + return self._config_data.networks[uuid] + + @locking.ssynchronized(_config_lock, shared=1) + def GetNetwork(self, uuid): + """Returns information about a network. + + It takes the information from the configuration file. + + @param uuid: UUID of the network + + @rtype: L{objects.Network} + @return: the network object + + """ + return self._UnlockedGetNetwork(uuid) + + @locking.ssynchronized(_config_lock) + def AddNetwork(self, net, ec_id, check_uuid=True): + """Add a network to the configuration. + + @type net: L{objects.Network} + @param net: the Network object to add + @type ec_id: string + @param ec_id: unique id for the job to use when creating a missing UUID + + """ + self._UnlockedAddNetwork(net, ec_id, check_uuid) + self._WriteConfig() + + def _UnlockedAddNetwork(self, net, ec_id, check_uuid): + """Add a network to the configuration. + + """ + logging.info("Adding network %s to configuration", net.name) + + if check_uuid: + self._EnsureUUID(net, ec_id) + + existing_uuid = self._UnlockedLookupNetwork(net.name) + if existing_uuid: + raise errors.OpPrereqError("Desired network name '%s' already" + " exists as a network (UUID: %s)" % + (net.name, existing_uuid), + errors.ECODE_EXISTS) + net.serial_no = 1 + self._config_data.networks[net.uuid] = net + self._config_data.cluster.serial_no += 1 + + def _UnlockedLookupNetwork(self, target): + """Lookup a network's UUID. + + @type target: string + @param target: network name or UUID + @rtype: string + @return: network UUID + @raises errors.OpPrereqError: when the target network cannot be found + + """ + if target in self._config_data.networks: + return target + for net in self._config_data.networks.values(): + if net.name == target: + return net.uuid + return None + + @locking.ssynchronized(_config_lock, shared=1) + def LookupNetwork(self, target): + """Lookup a network's UUID. + + This function is just a wrapper over L{_UnlockedLookupNetwork}. + + @type target: string + @param target: network name or UUID + @rtype: string + @return: network UUID + + """ + return self._UnlockedLookupNetwork(target) + + @locking.ssynchronized(_config_lock) + def RemoveNetwork(self, network_uuid): + """Remove a network from the configuration. + + @type network_uuid: string + @param network_uuid: the UUID of the network to remove + + """ + logging.info("Removing network %s from configuration", network_uuid) + + if network_uuid not in self._config_data.networks: + raise errors.ConfigurationError("Unknown network '%s'" % network_uuid) + + del self._config_data.networks[network_uuid] + self._config_data.cluster.serial_no += 1 + self._WriteConfig() + + def _UnlockedGetGroupNetParams(self, net, node): + """Get the netparams (mode, link) of a network. + + Get a network's netparams for a given node. + + @type net: string + @param net: network name + @type node: string + @param node: node name + @rtype: dict or None + @return: netparams + + """ + net_uuid = self._UnlockedLookupNetwork(net) + if net_uuid is None: + return None + + node_info = self._UnlockedGetNodeInfo(node) + nodegroup_info = self._UnlockedGetNodeGroup(node_info.group) + netparams = nodegroup_info.networks.get(net_uuid, None) + + return netparams + + @locking.ssynchronized(_config_lock, shared=1) + def GetGroupNetParams(self, net, node): + """Locking wrapper of _UnlockedGetGroupNetParams() + + """ + return self._UnlockedGetGroupNetParams(net, node) + + @locking.ssynchronized(_config_lock, shared=1) + def CheckIPInNodeGroup(self, ip, node): + """Check IP uniqueness in nodegroup. + + Check networks that are connected in the node's node group + if ip is contained in any of them. Used when creating/adding + a NIC to ensure uniqueness among nodegroups. + + @type ip: string + @param ip: ip address + @type node: string + @param node: node name + @rtype: (string, dict) or (None, None) + @return: (network name, netparams) + + """ + if ip is None: + return (None, None) + node_info = self._UnlockedGetNodeInfo(node) + nodegroup_info = self._UnlockedGetNodeGroup(node_info.group) + for net_uuid in nodegroup_info.networks.keys(): + net_info = self._UnlockedGetNetwork(net_uuid) + pool = network.AddressPool(net_info) + if pool.Contains(ip): + return (net_info.name, nodegroup_info.networks[net_uuid]) + + return (None, None)