code.grnet.gr Git - ganeti-local/blob - lib/config.py

   1 #
   2 #
   3
   4 # Copyright (C) 2006, 2007 Google Inc.
   5 #
   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.
  10 #
  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.
  15 #
  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
  19 # 02110-1301, USA.
  20
  21
  22 """Configuration management for Ganeti
  23
  24 This module provides the interface to the Ganeti cluster configuration.
  25
  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.
  28
  29 Currently, the data storage format is JSON. YAML was slow and consuming too
  30 much memory.
  31
  32 """
  33
  34 import os
  35 import tempfile
  36 import random
  37 import logging
  38
  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
  46
  47
  48 _config_lock = locking.SharedLock()
  49
  50
  51 def _ValidateConfig(data):
  52   if data.version != constants.CONFIG_VERSION:
  53     raise errors.ConfigurationError("Cluster configuration version"
  54                                     " mismatch, got %s instead of %s" %
  55                                     (data.version,
  56                                      constants.CONFIG_VERSION))
  57
  58
  59 class ConfigWriter:
  60   """The interface to the cluster configuration.
  61
  62   """
  63   def __init__(self, cfg_file=None, offline=False):
  64     self.write_count = 0
  65     self._lock = _config_lock
  66     self._config_data = None
  67     self._offline = offline
  68     if cfg_file is None:
  69       self._cfg_file = constants.CLUSTER_CONF_FILE
  70     else:
  71       self._cfg_file = cfg_file
  72     self._temporary_ids = set()
  73     self._temporary_drbds = {}
  74     # Note: in order to prevent errors when resolving our name in
  75     # _DistributeConfig, we compute it here once and reuse it; it's
  76     # better to raise an error before starting to modify the config
  77     # file than after it was modified
  78     self._my_hostname = utils.HostInfo().name
  79     self._OpenConfig()
  80
  81   # this method needs to be static, so that we can call it on the class
  82   @staticmethod
  83   def IsCluster():
  84     """Check if the cluster is configured.
  85
  86     """
  87     return os.path.exists(constants.CLUSTER_CONF_FILE)
  88
  89   @locking.ssynchronized(_config_lock, shared=1)
  90   def GenerateMAC(self):
  91     """Generate a MAC for an instance.
  92
  93     This should check the current instances for duplicates.
  94
  95     """
  96     prefix = self._config_data.cluster.mac_prefix
  97     all_macs = self._AllMACs()
  98     retries = 64
  99     while retries > 0:
 100       byte1 = random.randrange(0, 256)
 101       byte2 = random.randrange(0, 256)
 102       byte3 = random.randrange(0, 256)
 103       mac = "%s:%02x:%02x:%02x" % (prefix, byte1, byte2, byte3)
 104       if mac not in all_macs:
 105         break
 106       retries -= 1
 107     else:
 108       raise errors.ConfigurationError("Can't generate unique MAC")
 109     return mac
 110
 111   @locking.ssynchronized(_config_lock, shared=1)
 112   def IsMacInUse(self, mac):
 113     """Predicate: check if the specified MAC is in use in the Ganeti cluster.
 114
 115     This only checks instances managed by this cluster, it does not
 116     check for potential collisions elsewhere.
 117
 118     """
 119     all_macs = self._AllMACs()
 120     return mac in all_macs
 121
 122   @locking.ssynchronized(_config_lock, shared=1)
 123   def GenerateDRBDSecret(self):
 124     """Generate a DRBD secret.
 125
 126     This checks the current disks for duplicates.
 127
 128     """
 129     all_secrets = self._AllDRBDSecrets()
 130     retries = 64
 131     while retries > 0:
 132       secret = utils.GenerateSecret()
 133       if secret not in all_secrets:
 134         break
 135       retries -= 1
 136     else:
 137       raise errors.ConfigurationError("Can't generate unique DRBD secret")
 138     return secret
 139
 140   def _ComputeAllLVs(self):
 141     """Compute the list of all LVs.
 142
 143     """
 144     lvnames = set()
 145     for instance in self._config_data.instances.values():
 146       node_data = instance.MapLVsByNode()
 147       for lv_list in node_data.values():
 148         lvnames.update(lv_list)
 149     return lvnames
 150
 151   @locking.ssynchronized(_config_lock, shared=1)
 152   def GenerateUniqueID(self, exceptions=None):
 153     """Generate an unique disk name.
 154
 155     This checks the current node, instances and disk names for
 156     duplicates.
 157
 158     Args:
 159       - exceptions: a list with some other names which should be checked
 160                     for uniqueness (used for example when you want to get
 161                     more than one id at one time without adding each one in
 162                     turn to the config file
 163
 164     Returns: the unique id as a string
 165
 166     """
 167     existing = set()
 168     existing.update(self._temporary_ids)
 169     existing.update(self._ComputeAllLVs())
 170     existing.update(self._config_data.instances.keys())
 171     existing.update(self._config_data.nodes.keys())
 172     if exceptions is not None:
 173       existing.update(exceptions)
 174     retries = 64
 175     while retries > 0:
 176       unique_id = utils.NewUUID()
 177       if unique_id not in existing and unique_id is not None:
 178         break
 179     else:
 180       raise errors.ConfigurationError("Not able generate an unique ID"
 181                                       " (last tried ID: %s" % unique_id)
 182     self._temporary_ids.add(unique_id)
 183     return unique_id
 184
 185   def _AllMACs(self):
 186     """Return all MACs present in the config.
 187
 188     """
 189     result = []
 190     for instance in self._config_data.instances.values():
 191       for nic in instance.nics:
 192         result.append(nic.mac)
 193
 194     return result
 195
 196   def _AllDRBDSecrets(self):
 197     """Return all DRBD secrets present in the config.
 198
 199     """
 200     def helper(disk, result):
 201       """Recursively gather secrets from this disk."""
 202       if disk.dev_type == constants.DT_DRBD8:
 203         result.append(disk.logical_id[5])
 204       if disk.children:
 205         for child in disk.children:
 206           helper(child, result)
 207
 208     result = []
 209     for instance in self._config_data.instances.values():
 210       for disk in instance.disks:
 211         helper(disk, result)
 212
 213     return result
 214
 215   @locking.ssynchronized(_config_lock, shared=1)
 216   def VerifyConfig(self):
 217     """Verify function.
 218
 219     """
 220     result = []
 221     seen_macs = []
 222     ports = {}
 223     data = self._config_data
 224     for instance_name in data.instances:
 225       instance = data.instances[instance_name]
 226       if instance.primary_node not in data.nodes:
 227         result.append("instance '%s' has invalid primary node '%s'" %
 228                       (instance_name, instance.primary_node))
 229       for snode in instance.secondary_nodes:
 230         if snode not in data.nodes:
 231           result.append("instance '%s' has invalid secondary node '%s'" %
 232                         (instance_name, snode))
 233       for idx, nic in enumerate(instance.nics):
 234         if nic.mac in seen_macs:
 235           result.append("instance '%s' has NIC %d mac %s duplicate" %
 236                         (instance_name, idx, nic.mac))
 237         else:
 238           seen_macs.append(nic.mac)
 239
 240       # gather the drbd ports for duplicate checks
 241       for dsk in instance.disks:
 242         if dsk.dev_type in constants.LDS_DRBD:
 243           tcp_port = dsk.logical_id[2]
 244           if tcp_port not in ports:
 245             ports[tcp_port] = []
 246           ports[tcp_port].append((instance.name, "drbd disk %s" % dsk.iv_name))
 247       # gather network port reservation
 248       net_port = getattr(instance, "network_port", None)
 249       if net_port is not None:
 250         if net_port not in ports:
 251           ports[net_port] = []
 252         ports[net_port].append((instance.name, "network port"))
 253
 254     # cluster-wide pool of free ports
 255     for free_port in data.cluster.tcpudp_port_pool:
 256       if free_port not in ports:
 257         ports[free_port] = []
 258       ports[free_port].append(("cluster", "port marked as free"))
 259
 260     # compute tcp/udp duplicate ports
 261     keys = ports.keys()
 262     keys.sort()
 263     for pnum in keys:
 264       pdata = ports[pnum]
 265       if len(pdata) > 1:
 266         txt = ", ".join(["%s/%s" % val for val in pdata])
 267         result.append("tcp/udp port %s has duplicates: %s" % (pnum, txt))
 268
 269     # highest used tcp port check
 270     if keys:
 271       if keys[-1] > data.cluster.highest_used_port:
 272         result.append("Highest used port mismatch, saved %s, computed %s" %
 273                       (data.cluster.highest_used_port, keys[-1]))
 274
 275     if not data.nodes[data.cluster.master_node].master_candidate:
 276       result.append("Master node is not a master candidate")
 277
 278     cp_size = data.cluster.candidate_pool_size
 279     num_c = 0
 280     for node in data.nodes.values():
 281       if node.master_candidate:
 282         num_c += 1
 283     if cp_size > num_c and num_c < len(data.nodes):
 284       result.append("Not enough master candidates: actual %d, desired %d,"
 285                     " %d total nodes" % (num_c, cp_size, len(data.nodes)))
 286
 287     return result
 288
 289   def _UnlockedSetDiskID(self, disk, node_name):
 290     """Convert the unique ID to the ID needed on the target nodes.
 291
 292     This is used only for drbd, which needs ip/port configuration.
 293
 294     The routine descends down and updates its children also, because
 295     this helps when the only the top device is passed to the remote
 296     node.
 297
 298     This function is for internal use, when the config lock is already held.
 299
 300     """
 301     if disk.children:
 302       for child in disk.children:
 303         self._UnlockedSetDiskID(child, node_name)
 304
 305     if disk.logical_id is None and disk.physical_id is not None:
 306       return
 307     if disk.dev_type == constants.LD_DRBD8:
 308       pnode, snode, port, pminor, sminor, secret = disk.logical_id
 309       if node_name not in (pnode, snode):
 310         raise errors.ConfigurationError("DRBD device not knowing node %s" %
 311                                         node_name)
 312       pnode_info = self._UnlockedGetNodeInfo(pnode)
 313       snode_info = self._UnlockedGetNodeInfo(snode)
 314       if pnode_info is None or snode_info is None:
 315         raise errors.ConfigurationError("Can't find primary or secondary node"
 316                                         " for %s" % str(disk))
 317       p_data = (pnode_info.secondary_ip, port)
 318       s_data = (snode_info.secondary_ip, port)
 319       if pnode == node_name:
 320         disk.physical_id = p_data + s_data + (pminor, secret)
 321       else: # it must be secondary, we tested above
 322         disk.physical_id = s_data + p_data + (sminor, secret)
 323     else:
 324       disk.physical_id = disk.logical_id
 325     return
 326
 327   @locking.ssynchronized(_config_lock)
 328   def SetDiskID(self, disk, node_name):
 329     """Convert the unique ID to the ID needed on the target nodes.
 330
 331     This is used only for drbd, which needs ip/port configuration.
 332
 333     The routine descends down and updates its children also, because
 334     this helps when the only the top device is passed to the remote
 335     node.
 336
 337     """
 338     return self._UnlockedSetDiskID(disk, node_name)
 339
 340   @locking.ssynchronized(_config_lock)
 341   def AddTcpUdpPort(self, port):
 342     """Adds a new port to the available port pool.
 343
 344     """
 345     if not isinstance(port, int):
 346       raise errors.ProgrammerError("Invalid type passed for port")
 347
 348     self._config_data.cluster.tcpudp_port_pool.add(port)
 349     self._WriteConfig()
 350
 351   @locking.ssynchronized(_config_lock, shared=1)
 352   def GetPortList(self):
 353     """Returns a copy of the current port list.
 354
 355     """
 356     return self._config_data.cluster.tcpudp_port_pool.copy()
 357
 358   @locking.ssynchronized(_config_lock)
 359   def AllocatePort(self):
 360     """Allocate a port.
 361
 362     The port will be taken from the available port pool or from the
 363     default port range (and in this case we increase
 364     highest_used_port).
 365
 366     """
 367     # If there are TCP/IP ports configured, we use them first.
 368     if self._config_data.cluster.tcpudp_port_pool:
 369       port = self._config_data.cluster.tcpudp_port_pool.pop()
 370     else:
 371       port = self._config_data.cluster.highest_used_port + 1
 372       if port >= constants.LAST_DRBD_PORT:
 373         raise errors.ConfigurationError("The highest used port is greater"
 374                                         " than %s. Aborting." %
 375                                         constants.LAST_DRBD_PORT)
 376       self._config_data.cluster.highest_used_port = port
 377
 378     self._WriteConfig()
 379     return port
 380
 381   def _ComputeDRBDMap(self, instance):
 382     """Compute the used DRBD minor/nodes.
 383
 384     Return: dictionary of node_name: dict of minor: instance_name. The
 385     returned dict will have all the nodes in it (even if with an empty
 386     list).
 387
 388     """
 389     def _AppendUsedPorts(instance_name, disk, used):
 390       if disk.dev_type == constants.LD_DRBD8 and len(disk.logical_id) >= 5:
 391         nodeA, nodeB, dummy, minorA, minorB = disk.logical_id[:5]
 392         for node, port in ((nodeA, minorA), (nodeB, minorB)):
 393           assert node in used, "Instance node not found in node list"
 394           if port in used[node]:
 395             raise errors.ProgrammerError("DRBD minor already used:"
 396                                          " %s/%s, %s/%s" %
 397                                          (node, port, instance_name,
 398                                           used[node][port]))
 399
 400           used[node][port] = instance_name
 401       if disk.children:
 402         for child in disk.children:
 403           _AppendUsedPorts(instance_name, child, used)
 404
 405     my_dict = dict((node, {}) for node in self._config_data.nodes)
 406     for (node, minor), instance in self._temporary_drbds.iteritems():
 407       my_dict[node][minor] = instance
 408     for instance in self._config_data.instances.itervalues():
 409       for disk in instance.disks:
 410         _AppendUsedPorts(instance.name, disk, my_dict)
 411     return my_dict
 412
 413   @locking.ssynchronized(_config_lock)
 414   def AllocateDRBDMinor(self, nodes, instance):
 415     """Allocate a drbd minor.
 416
 417     The free minor will be automatically computed from the existing
 418     devices. A node can be given multiple times in order to allocate
 419     multiple minors. The result is the list of minors, in the same
 420     order as the passed nodes.
 421
 422     """
 423     d_map = self._ComputeDRBDMap(instance)
 424     result = []
 425     for nname in nodes:
 426       ndata = d_map[nname]
 427       if not ndata:
 428         # no minors used, we can start at 0
 429         result.append(0)
 430         ndata[0] = instance
 431         self._temporary_drbds[(nname, 0)] = instance
 432         continue
 433       keys = ndata.keys()
 434       keys.sort()
 435       ffree = utils.FirstFree(keys)
 436       if ffree is None:
 437         # return the next minor
 438         # TODO: implement high-limit check
 439         minor = keys[-1] + 1
 440       else:
 441         minor = ffree
 442       result.append(minor)
 443       ndata[minor] = instance
 444       assert (nname, minor) not in self._temporary_drbds, \
 445              "Attempt to reuse reserved DRBD minor"
 446       self._temporary_drbds[(nname, minor)] = instance
 447     logging.debug("Request to allocate drbd minors, input: %s, returning %s",
 448                   nodes, result)
 449     return result
 450
 451   @locking.ssynchronized(_config_lock)
 452   def ReleaseDRBDMinors(self, instance):
 453     """Release temporary drbd minors allocated for a given instance.
 454
 455     This should be called on both the error paths and on the success
 456     paths (after the instance has been added or updated).
 457
 458     @type instance: string
 459     @param instance: the instance for which temporary minors should be
 460                      released
 461
 462     """
 463     for key, name in self._temporary_drbds.items():
 464       if name == instance:
 465         del self._temporary_drbds[key]
 466
 467   @locking.ssynchronized(_config_lock, shared=1)
 468   def GetConfigVersion(self):
 469     """Get the configuration version.
 470
 471     @return: Config version
 472
 473     """
 474     return self._config_data.version
 475
 476   @locking.ssynchronized(_config_lock, shared=1)
 477   def GetClusterName(self):
 478     """Get cluster name.
 479
 480     @return: Cluster name
 481
 482     """
 483     return self._config_data.cluster.cluster_name
 484
 485   @locking.ssynchronized(_config_lock, shared=1)
 486   def GetMasterNode(self):
 487     """Get the hostname of the master node for this cluster.
 488
 489     @return: Master hostname
 490
 491     """
 492     return self._config_data.cluster.master_node
 493
 494   @locking.ssynchronized(_config_lock, shared=1)
 495   def GetMasterIP(self):
 496     """Get the IP of the master node for this cluster.
 497
 498     @return: Master IP
 499
 500     """
 501     return self._config_data.cluster.master_ip
 502
 503   @locking.ssynchronized(_config_lock, shared=1)
 504   def GetMasterNetdev(self):
 505     """Get the master network device for this cluster.
 506
 507     """
 508     return self._config_data.cluster.master_netdev
 509
 510   @locking.ssynchronized(_config_lock, shared=1)
 511   def GetFileStorageDir(self):
 512     """Get the file storage dir for this cluster.
 513
 514     """
 515     return self._config_data.cluster.file_storage_dir
 516
 517   @locking.ssynchronized(_config_lock, shared=1)
 518   def GetHypervisorType(self):
 519     """Get the hypervisor type for this cluster.
 520
 521     """
 522     return self._config_data.cluster.default_hypervisor
 523
 524   @locking.ssynchronized(_config_lock, shared=1)
 525   def GetHostKey(self):
 526     """Return the rsa hostkey from the config.
 527
 528     Args: None
 529
 530     Returns: rsa hostkey
 531     """
 532     return self._config_data.cluster.rsahostkeypub
 533
 534   @locking.ssynchronized(_config_lock)
 535   def AddInstance(self, instance):
 536     """Add an instance to the config.
 537
 538     This should be used after creating a new instance.
 539
 540     Args:
 541       instance: the instance object
 542     """
 543     if not isinstance(instance, objects.Instance):
 544       raise errors.ProgrammerError("Invalid type passed to AddInstance")
 545
 546     if instance.disk_template != constants.DT_DISKLESS:
 547       all_lvs = instance.MapLVsByNode()
 548       logging.info("Instance '%s' DISK_LAYOUT: %s", instance.name, all_lvs)
 549
 550     instance.serial_no = 1
 551     self._config_data.instances[instance.name] = instance
 552     self._WriteConfig()
 553
 554   def _SetInstanceStatus(self, instance_name, status):
 555     """Set the instance's status to a given value.
 556
 557     """
 558     if status not in ("up", "down"):
 559       raise errors.ProgrammerError("Invalid status '%s' passed to"
 560                                    " ConfigWriter._SetInstanceStatus()" %
 561                                    status)
 562
 563     if instance_name not in self._config_data.instances:
 564       raise errors.ConfigurationError("Unknown instance '%s'" %
 565                                       instance_name)
 566     instance = self._config_data.instances[instance_name]
 567     if instance.status != status:
 568       instance.status = status
 569       instance.serial_no += 1
 570       self._WriteConfig()
 571
 572   @locking.ssynchronized(_config_lock)
 573   def MarkInstanceUp(self, instance_name):
 574     """Mark the instance status to up in the config.
 575
 576     """
 577     self._SetInstanceStatus(instance_name, "up")
 578
 579   @locking.ssynchronized(_config_lock)
 580   def RemoveInstance(self, instance_name):
 581     """Remove the instance from the configuration.
 582
 583     """
 584     if instance_name not in self._config_data.instances:
 585       raise errors.ConfigurationError("Unknown instance '%s'" % instance_name)
 586     del self._config_data.instances[instance_name]
 587     self._WriteConfig()
 588
 589   @locking.ssynchronized(_config_lock)
 590   def RenameInstance(self, old_name, new_name):
 591     """Rename an instance.
 592
 593     This needs to be done in ConfigWriter and not by RemoveInstance
 594     combined with AddInstance as only we can guarantee an atomic
 595     rename.
 596
 597     """
 598     if old_name not in self._config_data.instances:
 599       raise errors.ConfigurationError("Unknown instance '%s'" % old_name)
 600     inst = self._config_data.instances[old_name]
 601     del self._config_data.instances[old_name]
 602     inst.name = new_name
 603
 604     for disk in inst.disks:
 605       if disk.dev_type == constants.LD_FILE:
 606         # rename the file paths in logical and physical id
 607         file_storage_dir = os.path.dirname(os.path.dirname(disk.logical_id[1]))
 608         disk.physical_id = disk.logical_id = (disk.logical_id[0],
 609                                               os.path.join(file_storage_dir,
 610                                                            inst.name,
 611                                                            disk.iv_name))
 612
 613     self._config_data.instances[inst.name] = inst
 614     self._WriteConfig()
 615
 616   @locking.ssynchronized(_config_lock)
 617   def MarkInstanceDown(self, instance_name):
 618     """Mark the status of an instance to down in the configuration.
 619
 620     """
 621     self._SetInstanceStatus(instance_name, "down")
 622
 623   def _UnlockedGetInstanceList(self):
 624     """Get the list of instances.
 625
 626     This function is for internal use, when the config lock is already held.
 627
 628     """
 629     return self._config_data.instances.keys()
 630
 631   @locking.ssynchronized(_config_lock, shared=1)
 632   def GetInstanceList(self):
 633     """Get the list of instances.
 634
 635     Returns:
 636       array of instances, ex. ['instance2.example.com','instance1.example.com']
 637       these contains all the instances, also the ones in Admin_down state
 638
 639     """
 640     return self._UnlockedGetInstanceList()
 641
 642   @locking.ssynchronized(_config_lock, shared=1)
 643   def ExpandInstanceName(self, short_name):
 644     """Attempt to expand an incomplete instance name.
 645
 646     """
 647     return utils.MatchNameComponent(short_name,
 648                                     self._config_data.instances.keys())
 649
 650   def _UnlockedGetInstanceInfo(self, instance_name):
 651     """Returns informations about an instance.
 652
 653     This function is for internal use, when the config lock is already held.
 654
 655     """
 656     if instance_name not in self._config_data.instances:
 657       return None
 658
 659     return self._config_data.instances[instance_name]
 660
 661   @locking.ssynchronized(_config_lock, shared=1)
 662   def GetInstanceInfo(self, instance_name):
 663     """Returns informations about an instance.
 664
 665     It takes the information from the configuration file. Other informations of
 666     an instance are taken from the live systems.
 667
 668     Args:
 669       instance: name of the instance, ex instance1.example.com
 670
 671     Returns:
 672       the instance object
 673
 674     """
 675     return self._UnlockedGetInstanceInfo(instance_name)
 676
 677   @locking.ssynchronized(_config_lock, shared=1)
 678   def GetAllInstancesInfo(self):
 679     """Get the configuration of all instances.
 680
 681     @rtype: dict
 682     @returns: dict of (instance, instance_info), where instance_info is what
 683               would GetInstanceInfo return for the node
 684
 685     """
 686     my_dict = dict([(instance, self._UnlockedGetInstanceInfo(instance))
 687                     for instance in self._UnlockedGetInstanceList()])
 688     return my_dict
 689
 690   @locking.ssynchronized(_config_lock)
 691   def AddNode(self, node):
 692     """Add a node to the configuration.
 693
 694     Args:
 695       node: an object.Node instance
 696
 697     """
 698     logging.info("Adding node %s to configuration" % node.name)
 699
 700     node.serial_no = 1
 701     self._config_data.nodes[node.name] = node
 702     self._config_data.cluster.serial_no += 1
 703     self._WriteConfig()
 704
 705   @locking.ssynchronized(_config_lock)
 706   def RemoveNode(self, node_name):
 707     """Remove a node from the configuration.
 708
 709     """
 710     logging.info("Removing node %s from configuration" % node_name)
 711
 712     if node_name not in self._config_data.nodes:
 713       raise errors.ConfigurationError("Unknown node '%s'" % node_name)
 714
 715     del self._config_data.nodes[node_name]
 716     self._config_data.cluster.serial_no += 1
 717     self._WriteConfig()
 718
 719   @locking.ssynchronized(_config_lock, shared=1)
 720   def ExpandNodeName(self, short_name):
 721     """Attempt to expand an incomplete instance name.
 722
 723     """
 724     return utils.MatchNameComponent(short_name,
 725                                     self._config_data.nodes.keys())
 726
 727   def _UnlockedGetNodeInfo(self, node_name):
 728     """Get the configuration of a node, as stored in the config.
 729
 730     This function is for internal use, when the config lock is already held.
 731
 732     Args: node: nodename (tuple) of the node
 733
 734     Returns: the node object
 735
 736     """
 737     if node_name not in self._config_data.nodes:
 738       return None
 739
 740     return self._config_data.nodes[node_name]
 741
 742
 743   @locking.ssynchronized(_config_lock, shared=1)
 744   def GetNodeInfo(self, node_name):
 745     """Get the configuration of a node, as stored in the config.
 746
 747     Args: node: nodename (tuple) of the node
 748
 749     Returns: the node object
 750
 751     """
 752     return self._UnlockedGetNodeInfo(node_name)
 753
 754   def _UnlockedGetNodeList(self):
 755     """Return the list of nodes which are in the configuration.
 756
 757     This function is for internal use, when the config lock is already held.
 758
 759     """
 760     return self._config_data.nodes.keys()
 761
 762
 763   @locking.ssynchronized(_config_lock, shared=1)
 764   def GetNodeList(self):
 765     """Return the list of nodes which are in the configuration.
 766
 767     """
 768     return self._UnlockedGetNodeList()
 769
 770   @locking.ssynchronized(_config_lock, shared=1)
 771   def GetAllNodesInfo(self):
 772     """Get the configuration of all nodes.
 773
 774     @rtype: dict
 775     @returns: dict of (node, node_info), where node_info is what
 776               would GetNodeInfo return for the node
 777
 778     """
 779     my_dict = dict([(node, self._UnlockedGetNodeInfo(node))
 780                     for node in self._UnlockedGetNodeList()])
 781     return my_dict
 782
 783   def _BumpSerialNo(self):
 784     """Bump up the serial number of the config.
 785
 786     """
 787     self._config_data.serial_no += 1
 788
 789   def _OpenConfig(self):
 790     """Read the config data from disk.
 791
 792     In case we already have configuration data and the config file has
 793     the same mtime as when we read it, we skip the parsing of the
 794     file, since de-serialisation could be slow.
 795
 796     """
 797     f = open(self._cfg_file, 'r')
 798     try:
 799       try:
 800         data = objects.ConfigData.FromDict(serializer.Load(f.read()))
 801       except Exception, err:
 802         raise errors.ConfigurationError(err)
 803     finally:
 804       f.close()
 805
 806     # Make sure the configuration has the right version
 807     _ValidateConfig(data)
 808
 809     if (not hasattr(data, 'cluster') or
 810         not hasattr(data.cluster, 'rsahostkeypub')):
 811       raise errors.ConfigurationError("Incomplete configuration"
 812                                       " (missing cluster.rsahostkeypub)")
 813     self._config_data = data
 814     # init the last serial as -1 so that the next write will cause
 815     # ssconf update
 816     self._last_cluster_serial = -1
 817
 818   def _DistributeConfig(self):
 819     """Distribute the configuration to the other nodes.
 820
 821     Currently, this only copies the configuration file. In the future,
 822     it could be used to encapsulate the 2/3-phase update mechanism.
 823
 824     """
 825     if self._offline:
 826       return True
 827     bad = False
 828
 829     node_list = []
 830     addr_list = []
 831     myhostname = self._my_hostname
 832     # we can skip checking whether _UnlockedGetNodeInfo returns None
 833     # since the node list comes from _UnlocketGetNodeList, and we are
 834     # called with the lock held, so no modifications should take place
 835     # in between
 836     for node_name in self._UnlockedGetNodeList():
 837       if node_name == myhostname:
 838         continue
 839       node_info = self._UnlockedGetNodeInfo(node_name)
 840       if not node_info.master_candidate:
 841         continue
 842       node_list.append(node_info.name)
 843       addr_list.append(node_info.primary_ip)
 844
 845     result = rpc.RpcRunner.call_upload_file(node_list, self._cfg_file,
 846                                             address_list=addr_list)
 847     for node in node_list:
 848       if not result[node]:
 849         logging.error("copy of file %s to node %s failed",
 850                       self._cfg_file, node)
 851         bad = True
 852     return not bad
 853
 854   def _WriteConfig(self, destination=None):
 855     """Write the configuration data to persistent storage.
 856
 857     """
 858     if destination is None:
 859       destination = self._cfg_file
 860     self._BumpSerialNo()
 861     txt = serializer.Dump(self._config_data.ToDict())
 862     dir_name, file_name = os.path.split(destination)
 863     fd, name = tempfile.mkstemp('.newconfig', file_name, dir_name)
 864     f = os.fdopen(fd, 'w')
 865     try:
 866       f.write(txt)
 867       os.fsync(f.fileno())
 868     finally:
 869       f.close()
 870     # we don't need to do os.close(fd) as f.close() did it
 871     os.rename(name, destination)
 872     self.write_count += 1
 873
 874     # and redistribute the config file to master candidates
 875     self._DistributeConfig()
 876
 877     # Write ssconf files on all nodes (including locally)
 878     if self._last_cluster_serial < self._config_data.cluster.serial_no:
 879       if not self._offline:
 880         rpc.RpcRunner.call_write_ssconf_files(self._UnlockedGetNodeList(),
 881                                               self._UnlockedGetSsconfValues())
 882       self._last_cluster_serial = self._config_data.cluster.serial_no
 883
 884   def _UnlockedGetSsconfValues(self):
 885     """Return the values needed by ssconf.
 886
 887     @rtype: dict
 888     @return: a dictionary with keys the ssconf names and values their
 889         associated value
 890
 891     """
 892     node_list = utils.NiceSort(self._UnlockedGetNodeList())
 893     mc_list = [self._UnlockedGetNodeInfo(name) for name in node_list]
 894     mc_list = [node.name for node in mc_list if node.master_candidate]
 895     node_list = "\n".join(node_list)
 896     mc_list = "\n".join(mc_list)
 897
 898     cluster = self._config_data.cluster
 899     return {
 900       constants.SS_CLUSTER_NAME: cluster.cluster_name,
 901       constants.SS_FILE_STORAGE_DIR: cluster.file_storage_dir,
 902       constants.SS_MASTER_CANDIDATES: mc_list,
 903       constants.SS_MASTER_IP: cluster.master_ip,
 904       constants.SS_MASTER_NETDEV: cluster.master_netdev,
 905       constants.SS_MASTER_NODE: cluster.master_node,
 906       constants.SS_NODE_LIST: node_list,
 907       }
 908
 909   @locking.ssynchronized(_config_lock)
 910   def InitConfig(self, version, cluster_config, master_node_config):
 911     """Create the initial cluster configuration.
 912
 913     It will contain the current node, which will also be the master
 914     node, and no instances.
 915
 916     @type version: int
 917     @param version: Configuration version
 918     @type cluster_config: objects.Cluster
 919     @param cluster_config: Cluster configuration
 920     @type master_node_config: objects.Node
 921     @param master_node_config: Master node configuration
 922
 923     """
 924     nodes = {
 925       master_node_config.name: master_node_config,
 926       }
 927
 928     self._config_data = objects.ConfigData(version=version,
 929                                            cluster=cluster_config,
 930                                            nodes=nodes,
 931                                            instances={},
 932                                            serial_no=1)
 933     self._WriteConfig()
 934
 935   @locking.ssynchronized(_config_lock, shared=1)
 936   def GetVGName(self):
 937     """Return the volume group name.
 938
 939     """
 940     return self._config_data.cluster.volume_group_name
 941
 942   @locking.ssynchronized(_config_lock)
 943   def SetVGName(self, vg_name):
 944     """Set the volume group name.
 945
 946     """
 947     self._config_data.cluster.volume_group_name = vg_name
 948     self._config_data.cluster.serial_no += 1
 949     self._WriteConfig()
 950
 951   @locking.ssynchronized(_config_lock, shared=1)
 952   def GetDefBridge(self):
 953     """Return the default bridge.
 954
 955     """
 956     return self._config_data.cluster.default_bridge
 957
 958   @locking.ssynchronized(_config_lock, shared=1)
 959   def GetMACPrefix(self):
 960     """Return the mac prefix.
 961
 962     """
 963     return self._config_data.cluster.mac_prefix
 964
 965   @locking.ssynchronized(_config_lock, shared=1)
 966   def GetClusterInfo(self):
 967     """Returns informations about the cluster
 968
 969     Returns:
 970       the cluster object
 971
 972     """
 973     return self._config_data.cluster
 974
 975   @locking.ssynchronized(_config_lock)
 976   def Update(self, target):
 977     """Notify function to be called after updates.
 978
 979     This function must be called when an object (as returned by
 980     GetInstanceInfo, GetNodeInfo, GetCluster) has been updated and the
 981     caller wants the modifications saved to the backing store. Note
 982     that all modified objects will be saved, but the target argument
 983     is the one the caller wants to ensure that it's saved.
 984
 985     """
 986     if self._config_data is None:
 987       raise errors.ProgrammerError("Configuration file not read,"
 988                                    " cannot save.")
 989     update_serial = False
 990     if isinstance(target, objects.Cluster):
 991       test = target == self._config_data.cluster
 992     elif isinstance(target, objects.Node):
 993       test = target in self._config_data.nodes.values()
 994       update_serial = True
 995     elif isinstance(target, objects.Instance):
 996       test = target in self._config_data.instances.values()
 997     else:
 998       raise errors.ProgrammerError("Invalid object type (%s) passed to"
 999                                    " ConfigWriter.Update" % type(target))
1000     if not test:
1001       raise errors.ConfigurationError("Configuration updated since object"
1002                                       " has been read or unknown object")
1003     target.serial_no += 1
1004
1005     if update_serial:
1006       # for node updates, we need to increase the cluster serial too
1007       self._config_data.cluster.serial_no += 1
1008
1009     self._WriteConfig()