X-Git-Url: https://code.grnet.gr/git/ganeti-local/blobdiff_plain/c79198a032e7663078cac8e8128a12d434170823..37e61a779d82a83555942bdf9b224a848a47ea46:/lib/config.py?ds=sidebyside diff --git a/lib/config.py b/lib/config.py index ee9747c..97b3939 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 @@ -38,6 +38,7 @@ 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(): @@ -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,8 +601,11 @@ 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) @@ -884,6 +954,13 @@ class ConfigWriter: 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. @@ -1155,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() @@ -1173,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): @@ -1182,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() @@ -1197,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) @@ -1222,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. @@ -1321,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. @@ -1625,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. @@ -1749,7 +1933,7 @@ class ConfigWriter: # TODO: Use dedicated resolver talking to config writer for name resolution result = \ - rpc.ConfigRunner(addr_list).call_upload_file(node_list, self._cfg_file) + 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: @@ -1808,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.ConfigRunner(None).call_write_ssconf_files( + result = self._GetRpc(None).call_write_ssconf_files( self._UnlockedGetOnlineNodeList(), self._UnlockedGetSsconfValues())