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     cp_size = data.cluster.candidate_pool_size
 276     num_c = 0
 277     for node in data.nodes.values():
 278       if node.master_candidate:
 279         num_c += 1
 280     if cp_size > num_c:
 281       result.append("Not enough master candidates: actual %d, desired %d" %
 282                     (num_c, cp_size))
 283
 284     return result
 285
 286   def _UnlockedSetDiskID(self, disk, node_name):
 287     """Convert the unique ID to the ID needed on the target nodes.
 288
 289     This is used only for drbd, which needs ip/port configuration.
 290
 291     The routine descends down and updates its children also, because
 292     this helps when the only the top device is passed to the remote
 293     node.
 294
 295     This function is for internal use, when the config lock is already held.
 296
 297     """
 298     if disk.children:
 299       for child in disk.children:
 300         self._UnlockedSetDiskID(child, node_name)
 301
 302     if disk.logical_id is None and disk.physical_id is not None:
 303       return
 304     if disk.dev_type == constants.LD_DRBD8:
 305       pnode, snode, port, pminor, sminor, secret = disk.logical_id
 306       if node_name not in (pnode, snode):
 307         raise errors.ConfigurationError("DRBD device not knowing node %s" %
 308                                         node_name)
 309       pnode_info = self._UnlockedGetNodeInfo(pnode)
 310       snode_info = self._UnlockedGetNodeInfo(snode)
 311       if pnode_info is None or snode_info is None:
 312         raise errors.ConfigurationError("Can't find primary or secondary node"
 313                                         " for %s" % str(disk))
 314       p_data = (pnode_info.secondary_ip, port)
 315       s_data = (snode_info.secondary_ip, port)
 316       if pnode == node_name:
 317         disk.physical_id = p_data + s_data + (pminor, secret)
 318       else: # it must be secondary, we tested above
 319         disk.physical_id = s_data + p_data + (sminor, secret)
 320     else:
 321       disk.physical_id = disk.logical_id
 322     return
 323
 324   @locking.ssynchronized(_config_lock)
 325   def SetDiskID(self, disk, node_name):
 326     """Convert the unique ID to the ID needed on the target nodes.
 327
 328     This is used only for drbd, which needs ip/port configuration.
 329
 330     The routine descends down and updates its children also, because
 331     this helps when the only the top device is passed to the remote
 332     node.
 333
 334     """
 335     return self._UnlockedSetDiskID(disk, node_name)
 336
 337   @locking.ssynchronized(_config_lock)
 338   def AddTcpUdpPort(self, port):
 339     """Adds a new port to the available port pool.
 340
 341     """
 342     if not isinstance(port, int):
 343       raise errors.ProgrammerError("Invalid type passed for port")
 344
 345     self._config_data.cluster.tcpudp_port_pool.add(port)
 346     self._WriteConfig()
 347
 348   @locking.ssynchronized(_config_lock, shared=1)
 349   def GetPortList(self):
 350     """Returns a copy of the current port list.
 351
 352     """
 353     return self._config_data.cluster.tcpudp_port_pool.copy()
 354
 355   @locking.ssynchronized(_config_lock)
 356   def AllocatePort(self):
 357     """Allocate a port.
 358
 359     The port will be taken from the available port pool or from the
 360     default port range (and in this case we increase
 361     highest_used_port).
 362
 363     """
 364     # If there are TCP/IP ports configured, we use them first.
 365     if self._config_data.cluster.tcpudp_port_pool:
 366       port = self._config_data.cluster.tcpudp_port_pool.pop()
 367     else:
 368       port = self._config_data.cluster.highest_used_port + 1
 369       if port >= constants.LAST_DRBD_PORT:
 370         raise errors.ConfigurationError("The highest used port is greater"
 371                                         " than %s. Aborting." %
 372                                         constants.LAST_DRBD_PORT)
 373       self._config_data.cluster.highest_used_port = port
 374
 375     self._WriteConfig()
 376     return port
 377
 378   def _ComputeDRBDMap(self, instance):
 379     """Compute the used DRBD minor/nodes.
 380
 381     Return: dictionary of node_name: dict of minor: instance_name. The
 382     returned dict will have all the nodes in it (even if with an empty
 383     list).
 384
 385     """
 386     def _AppendUsedPorts(instance_name, disk, used):
 387       if disk.dev_type == constants.LD_DRBD8 and len(disk.logical_id) >= 5:
 388         nodeA, nodeB, dummy, minorA, minorB = disk.logical_id[:5]
 389         for node, port in ((nodeA, minorA), (nodeB, minorB)):
 390           assert node in used, "Instance node not found in node list"
 391           if port in used[node]:
 392             raise errors.ProgrammerError("DRBD minor already used:"
 393                                          " %s/%s, %s/%s" %
 394                                          (node, port, instance_name,
 395                                           used[node][port]))
 396
 397           used[node][port] = instance_name
 398       if disk.children:
 399         for child in disk.children:
 400           _AppendUsedPorts(instance_name, child, used)
 401
 402     my_dict = dict((node, {}) for node in self._config_data.nodes)
 403     for (node, minor), instance in self._temporary_drbds.iteritems():
 404       my_dict[node][minor] = instance
 405     for instance in self._config_data.instances.itervalues():
 406       for disk in instance.disks:
 407         _AppendUsedPorts(instance.name, disk, my_dict)
 408     return my_dict
 409
 410   @locking.ssynchronized(_config_lock)
 411   def AllocateDRBDMinor(self, nodes, instance):
 412     """Allocate a drbd minor.
 413
 414     The free minor will be automatically computed from the existing
 415     devices. A node can be given multiple times in order to allocate
 416     multiple minors. The result is the list of minors, in the same
 417     order as the passed nodes.
 418
 419     """
 420     d_map = self._ComputeDRBDMap(instance)
 421     result = []
 422     for nname in nodes:
 423       ndata = d_map[nname]
 424       if not ndata:
 425         # no minors used, we can start at 0
 426         result.append(0)
 427         ndata[0] = instance
 428         self._temporary_drbds[(nname, 0)] = instance
 429         continue
 430       keys = ndata.keys()
 431       keys.sort()
 432       ffree = utils.FirstFree(keys)
 433       if ffree is None:
 434         # return the next minor
 435         # TODO: implement high-limit check
 436         minor = keys[-1] + 1
 437       else:
 438         minor = ffree
 439       result.append(minor)
 440       ndata[minor] = instance
 441       assert (nname, minor) not in self._temporary_drbds, \
 442              "Attempt to reuse reserved DRBD minor"
 443       self._temporary_drbds[(nname, minor)] = instance
 444     logging.debug("Request to allocate drbd minors, input: %s, returning %s",
 445                   nodes, result)
 446     return result
 447
 448   @locking.ssynchronized(_config_lock)
 449   def ReleaseDRBDMinors(self, instance):
 450     """Release temporary drbd minors allocated for a given instance.
 451
 452     This should be called on both the error paths and on the success
 453     paths (after the instance has been added or updated).
 454
 455     @type instance: string
 456     @param instance: the instance for which temporary minors should be
 457                      released
 458
 459     """
 460     for key, name in self._temporary_drbds.items():
 461       if name == instance:
 462         del self._temporary_drbds[key]
 463
 464   @locking.ssynchronized(_config_lock, shared=1)
 465   def GetConfigVersion(self):
 466     """Get the configuration version.
 467
 468     @return: Config version
 469
 470     """
 471     return self._config_data.version
 472
 473   @locking.ssynchronized(_config_lock, shared=1)
 474   def GetClusterName(self):
 475     """Get cluster name.
 476
 477     @return: Cluster name
 478
 479     """
 480     return self._config_data.cluster.cluster_name
 481
 482   @locking.ssynchronized(_config_lock, shared=1)
 483   def GetMasterNode(self):
 484     """Get the hostname of the master node for this cluster.
 485
 486     @return: Master hostname
 487
 488     """
 489     return self._config_data.cluster.master_node
 490
 491   @locking.ssynchronized(_config_lock, shared=1)
 492   def GetMasterIP(self):
 493     """Get the IP of the master node for this cluster.
 494
 495     @return: Master IP
 496
 497     """
 498     return self._config_data.cluster.master_ip
 499
 500   @locking.ssynchronized(_config_lock, shared=1)
 501   def GetMasterNetdev(self):
 502     """Get the master network device for this cluster.
 503
 504     """
 505     return self._config_data.cluster.master_netdev
 506
 507   @locking.ssynchronized(_config_lock, shared=1)
 508   def GetFileStorageDir(self):
 509     """Get the file storage dir for this cluster.
 510
 511     """
 512     return self._config_data.cluster.file_storage_dir
 513
 514   @locking.ssynchronized(_config_lock, shared=1)
 515   def GetHypervisorType(self):
 516     """Get the hypervisor type for this cluster.
 517
 518     """
 519     return self._config_data.cluster.default_hypervisor
 520
 521   @locking.ssynchronized(_config_lock, shared=1)
 522   def GetHostKey(self):
 523     """Return the rsa hostkey from the config.
 524
 525     Args: None
 526
 527     Returns: rsa hostkey
 528     """
 529     return self._config_data.cluster.rsahostkeypub
 530
 531   @locking.ssynchronized(_config_lock)
 532   def AddInstance(self, instance):
 533     """Add an instance to the config.
 534
 535     This should be used after creating a new instance.
 536
 537     Args:
 538       instance: the instance object
 539     """
 540     if not isinstance(instance, objects.Instance):
 541       raise errors.ProgrammerError("Invalid type passed to AddInstance")
 542
 543     if instance.disk_template != constants.DT_DISKLESS:
 544       all_lvs = instance.MapLVsByNode()
 545       logging.info("Instance '%s' DISK_LAYOUT: %s", instance.name, all_lvs)
 546
 547     instance.serial_no = 1
 548     self._config_data.instances[instance.name] = instance
 549     self._WriteConfig()
 550
 551   def _SetInstanceStatus(self, instance_name, status):
 552     """Set the instance's status to a given value.
 553
 554     """
 555     if status not in ("up", "down"):
 556       raise errors.ProgrammerError("Invalid status '%s' passed to"
 557                                    " ConfigWriter._SetInstanceStatus()" %
 558                                    status)
 559
 560     if instance_name not in self._config_data.instances:
 561       raise errors.ConfigurationError("Unknown instance '%s'" %
 562                                       instance_name)
 563     instance = self._config_data.instances[instance_name]
 564     if instance.status != status:
 565       instance.status = status
 566       instance.serial_no += 1
 567       self._WriteConfig()
 568
 569   @locking.ssynchronized(_config_lock)
 570   def MarkInstanceUp(self, instance_name):
 571     """Mark the instance status to up in the config.
 572
 573     """
 574     self._SetInstanceStatus(instance_name, "up")
 575
 576   @locking.ssynchronized(_config_lock)
 577   def RemoveInstance(self, instance_name):
 578     """Remove the instance from the configuration.
 579
 580     """
 581     if instance_name not in self._config_data.instances:
 582       raise errors.ConfigurationError("Unknown instance '%s'" % instance_name)
 583     del self._config_data.instances[instance_name]
 584     self._WriteConfig()
 585
 586   @locking.ssynchronized(_config_lock)
 587   def RenameInstance(self, old_name, new_name):
 588     """Rename an instance.
 589
 590     This needs to be done in ConfigWriter and not by RemoveInstance
 591     combined with AddInstance as only we can guarantee an atomic
 592     rename.
 593
 594     """
 595     if old_name not in self._config_data.instances:
 596       raise errors.ConfigurationError("Unknown instance '%s'" % old_name)
 597     inst = self._config_data.instances[old_name]
 598     del self._config_data.instances[old_name]
 599     inst.name = new_name
 600
 601     for disk in inst.disks:
 602       if disk.dev_type == constants.LD_FILE:
 603         # rename the file paths in logical and physical id
 604         file_storage_dir = os.path.dirname(os.path.dirname(disk.logical_id[1]))
 605         disk.physical_id = disk.logical_id = (disk.logical_id[0],
 606                                               os.path.join(file_storage_dir,
 607                                                            inst.name,
 608                                                            disk.iv_name))
 609
 610     self._config_data.instances[inst.name] = inst
 611     self._WriteConfig()
 612
 613   @locking.ssynchronized(_config_lock)
 614   def MarkInstanceDown(self, instance_name):
 615     """Mark the status of an instance to down in the configuration.
 616
 617     """
 618     self._SetInstanceStatus(instance_name, "down")
 619
 620   def _UnlockedGetInstanceList(self):
 621     """Get the list of instances.
 622
 623     This function is for internal use, when the config lock is already held.
 624
 625     """
 626     return self._config_data.instances.keys()
 627
 628   @locking.ssynchronized(_config_lock, shared=1)
 629   def GetInstanceList(self):
 630     """Get the list of instances.
 631
 632     Returns:
 633       array of instances, ex. ['instance2.example.com','instance1.example.com']
 634       these contains all the instances, also the ones in Admin_down state
 635
 636     """
 637     return self._UnlockedGetInstanceList()
 638
 639   @locking.ssynchronized(_config_lock, shared=1)
 640   def ExpandInstanceName(self, short_name):
 641     """Attempt to expand an incomplete instance name.
 642
 643     """
 644     return utils.MatchNameComponent(short_name,
 645                                     self._config_data.instances.keys())
 646
 647   def _UnlockedGetInstanceInfo(self, instance_name):
 648     """Returns informations about an instance.
 649
 650     This function is for internal use, when the config lock is already held.
 651
 652     """
 653     if instance_name not in self._config_data.instances:
 654       return None
 655
 656     return self._config_data.instances[instance_name]
 657
 658   @locking.ssynchronized(_config_lock, shared=1)
 659   def GetInstanceInfo(self, instance_name):
 660     """Returns informations about an instance.
 661
 662     It takes the information from the configuration file. Other informations of
 663     an instance are taken from the live systems.
 664
 665     Args:
 666       instance: name of the instance, ex instance1.example.com
 667
 668     Returns:
 669       the instance object
 670
 671     """
 672     return self._UnlockedGetInstanceInfo(instance_name)
 673
 674   @locking.ssynchronized(_config_lock, shared=1)
 675   def GetAllInstancesInfo(self):
 676     """Get the configuration of all instances.
 677
 678     @rtype: dict
 679     @returns: dict of (instance, instance_info), where instance_info is what
 680               would GetInstanceInfo return for the node
 681
 682     """
 683     my_dict = dict([(instance, self._UnlockedGetInstanceInfo(instance))
 684                     for instance in self._UnlockedGetInstanceList()])
 685     return my_dict
 686
 687   @locking.ssynchronized(_config_lock)
 688   def AddNode(self, node):
 689     """Add a node to the configuration.
 690
 691     Args:
 692       node: an object.Node instance
 693
 694     """
 695     logging.info("Adding node %s to configuration" % node.name)
 696
 697     node.serial_no = 1
 698     self._config_data.nodes[node.name] = node
 699     self._config_data.cluster.serial_no += 1
 700     self._WriteConfig()
 701
 702   @locking.ssynchronized(_config_lock)
 703   def RemoveNode(self, node_name):
 704     """Remove a node from the configuration.
 705
 706     """
 707     logging.info("Removing node %s from configuration" % node_name)
 708
 709     if node_name not in self._config_data.nodes:
 710       raise errors.ConfigurationError("Unknown node '%s'" % node_name)
 711
 712     del self._config_data.nodes[node_name]
 713     self._config_data.cluster.serial_no += 1
 714     self._WriteConfig()
 715
 716   @locking.ssynchronized(_config_lock, shared=1)
 717   def ExpandNodeName(self, short_name):
 718     """Attempt to expand an incomplete instance name.
 719
 720     """
 721     return utils.MatchNameComponent(short_name,
 722                                     self._config_data.nodes.keys())
 723
 724   def _UnlockedGetNodeInfo(self, node_name):
 725     """Get the configuration of a node, as stored in the config.
 726
 727     This function is for internal use, when the config lock is already held.
 728
 729     Args: node: nodename (tuple) of the node
 730
 731     Returns: the node object
 732
 733     """
 734     if node_name not in self._config_data.nodes:
 735       return None
 736
 737     return self._config_data.nodes[node_name]
 738
 739
 740   @locking.ssynchronized(_config_lock, shared=1)
 741   def GetNodeInfo(self, node_name):
 742     """Get the configuration of a node, as stored in the config.
 743
 744     Args: node: nodename (tuple) of the node
 745
 746     Returns: the node object
 747
 748     """
 749     return self._UnlockedGetNodeInfo(node_name)
 750
 751   def _UnlockedGetNodeList(self):
 752     """Return the list of nodes which are in the configuration.
 753
 754     This function is for internal use, when the config lock is already held.
 755
 756     """
 757     return self._config_data.nodes.keys()
 758
 759
 760   @locking.ssynchronized(_config_lock, shared=1)
 761   def GetNodeList(self):
 762     """Return the list of nodes which are in the configuration.
 763
 764     """
 765     return self._UnlockedGetNodeList()
 766
 767   @locking.ssynchronized(_config_lock, shared=1)
 768   def GetAllNodesInfo(self):
 769     """Get the configuration of all nodes.
 770
 771     @rtype: dict
 772     @returns: dict of (node, node_info), where node_info is what
 773               would GetNodeInfo return for the node
 774
 775     """
 776     my_dict = dict([(node, self._UnlockedGetNodeInfo(node))
 777                     for node in self._UnlockedGetNodeList()])
 778     return my_dict
 779
 780   def _BumpSerialNo(self):
 781     """Bump up the serial number of the config.
 782
 783     """
 784     self._config_data.serial_no += 1
 785
 786   def _OpenConfig(self):
 787     """Read the config data from disk.
 788
 789     In case we already have configuration data and the config file has
 790     the same mtime as when we read it, we skip the parsing of the
 791     file, since de-serialisation could be slow.
 792
 793     """
 794     f = open(self._cfg_file, 'r')
 795     try:
 796       try:
 797         data = objects.ConfigData.FromDict(serializer.Load(f.read()))
 798       except Exception, err:
 799         raise errors.ConfigurationError(err)
 800     finally:
 801       f.close()
 802
 803     # Make sure the configuration has the right version
 804     _ValidateConfig(data)
 805
 806     if (not hasattr(data, 'cluster') or
 807         not hasattr(data.cluster, 'rsahostkeypub')):
 808       raise errors.ConfigurationError("Incomplete configuration"
 809                                       " (missing cluster.rsahostkeypub)")
 810     self._config_data = data
 811     # init the last serial as -1 so that the next write will cause
 812     # ssconf update
 813     self._last_cluster_serial = -1
 814
 815   def _DistributeConfig(self):
 816     """Distribute the configuration to the other nodes.
 817
 818     Currently, this only copies the configuration file. In the future,
 819     it could be used to encapsulate the 2/3-phase update mechanism.
 820
 821     """
 822     if self._offline:
 823       return True
 824     bad = False
 825
 826     node_list = []
 827     addr_list = []
 828     myhostname = self._my_hostname
 829     # we can skip checking whether _UnlockedGetNodeInfo returns None
 830     # since the node list comes from _UnlocketGetNodeList, and we are
 831     # called with the lock held, so no modifications should take place
 832     # in between
 833     for node_name in self._UnlockedGetNodeList():
 834       if node_name == myhostname:
 835         continue
 836       node_info = self._UnlockedGetNodeInfo(node_name)
 837       if not node_info.master_candidate:
 838         continue
 839       node_list.append(node_info.name)
 840       addr_list.append(node_info.primary_ip)
 841
 842     result = rpc.RpcRunner.call_upload_file(node_list, self._cfg_file,
 843                                             address_list=addr_list)
 844     for node in node_list:
 845       if not result[node]:
 846         logging.error("copy of file %s to node %s failed",
 847                       self._cfg_file, node)
 848         bad = True
 849     return not bad
 850
 851   def _WriteConfig(self, destination=None):
 852     """Write the configuration data to persistent storage.
 853
 854     """
 855     if destination is None:
 856       destination = self._cfg_file
 857     self._BumpSerialNo()
 858     txt = serializer.Dump(self._config_data.ToDict())
 859     dir_name, file_name = os.path.split(destination)
 860     fd, name = tempfile.mkstemp('.newconfig', file_name, dir_name)
 861     f = os.fdopen(fd, 'w')
 862     try:
 863       f.write(txt)
 864       os.fsync(f.fileno())
 865     finally:
 866       f.close()
 867     # we don't need to do os.close(fd) as f.close() did it
 868     os.rename(name, destination)
 869     self.write_count += 1
 870
 871     # and redistribute the config file to master candidates
 872     self._DistributeConfig()
 873
 874     # Write ssconf files on all nodes (including locally)
 875     if self._last_cluster_serial < self._config_data.cluster.serial_no:
 876       if not self._offline:
 877         rpc.RpcRunner.call_write_ssconf_files(self._UnlockedGetNodeList(),
 878                                               self._UnlockedGetSsconfValues())
 879       self._last_cluster_serial = self._config_data.cluster.serial_no
 880
 881   def _UnlockedGetSsconfValues(self):
 882     """Return the values needed by ssconf.
 883
 884     @rtype: dict
 885     @return: a dictionary with keys the ssconf names and values their
 886         associated value
 887
 888     """
 889     node_list = utils.NiceSort(self._UnlockedGetNodeList())
 890     mc_list = [self._UnlockedGetNodeInfo(name) for name in node_list]
 891     mc_list = [node.name for node in mc_list if node.master_candidate]
 892     node_list = "\n".join(node_list)
 893     mc_list = "\n".join(mc_list)
 894
 895     cluster = self._config_data.cluster
 896     return {
 897       constants.SS_CLUSTER_NAME: cluster.cluster_name,
 898       constants.SS_FILE_STORAGE_DIR: cluster.file_storage_dir,
 899       constants.SS_MASTER_CANDIDATES: mc_list,
 900       constants.SS_MASTER_IP: cluster.master_ip,
 901       constants.SS_MASTER_NETDEV: cluster.master_netdev,
 902       constants.SS_MASTER_NODE: cluster.master_node,
 903       constants.SS_NODE_LIST: node_list,
 904       }
 905
 906   @locking.ssynchronized(_config_lock)
 907   def InitConfig(self, version, cluster_config, master_node_config):
 908     """Create the initial cluster configuration.
 909
 910     It will contain the current node, which will also be the master
 911     node, and no instances.
 912
 913     @type version: int
 914     @param version: Configuration version
 915     @type cluster_config: objects.Cluster
 916     @param cluster_config: Cluster configuration
 917     @type master_node_config: objects.Node
 918     @param master_node_config: Master node configuration
 919
 920     """
 921     nodes = {
 922       master_node_config.name: master_node_config,
 923       }
 924
 925     self._config_data = objects.ConfigData(version=version,
 926                                            cluster=cluster_config,
 927                                            nodes=nodes,
 928                                            instances={},
 929                                            serial_no=1)
 930     self._WriteConfig()
 931
 932   @locking.ssynchronized(_config_lock, shared=1)
 933   def GetVGName(self):
 934     """Return the volume group name.
 935
 936     """
 937     return self._config_data.cluster.volume_group_name
 938
 939   @locking.ssynchronized(_config_lock)
 940   def SetVGName(self, vg_name):
 941     """Set the volume group name.
 942
 943     """
 944     self._config_data.cluster.volume_group_name = vg_name
 945     self._config_data.cluster.serial_no += 1
 946     self._WriteConfig()
 947
 948   @locking.ssynchronized(_config_lock, shared=1)
 949   def GetDefBridge(self):
 950     """Return the default bridge.
 951
 952     """
 953     return self._config_data.cluster.default_bridge
 954
 955   @locking.ssynchronized(_config_lock, shared=1)
 956   def GetMACPrefix(self):
 957     """Return the mac prefix.
 958
 959     """
 960     return self._config_data.cluster.mac_prefix
 961
 962   @locking.ssynchronized(_config_lock, shared=1)
 963   def GetClusterInfo(self):
 964     """Returns informations about the cluster
 965
 966     Returns:
 967       the cluster object
 968
 969     """
 970     return self._config_data.cluster
 971
 972   @locking.ssynchronized(_config_lock)
 973   def Update(self, target):
 974     """Notify function to be called after updates.
 975
 976     This function must be called when an object (as returned by
 977     GetInstanceInfo, GetNodeInfo, GetCluster) has been updated and the
 978     caller wants the modifications saved to the backing store. Note
 979     that all modified objects will be saved, but the target argument
 980     is the one the caller wants to ensure that it's saved.
 981
 982     """
 983     if self._config_data is None:
 984       raise errors.ProgrammerError("Configuration file not read,"
 985                                    " cannot save.")
 986     update_serial = False
 987     if isinstance(target, objects.Cluster):
 988       test = target == self._config_data.cluster
 989     elif isinstance(target, objects.Node):
 990       test = target in self._config_data.nodes.values()
 991       update_serial = True
 992     elif isinstance(target, objects.Instance):
 993       test = target in self._config_data.instances.values()
 994     else:
 995       raise errors.ProgrammerError("Invalid object type (%s) passed to"
 996                                    " ConfigWriter.Update" % type(target))
 997     if not test:
 998       raise errors.ConfigurationError("Configuration updated since object"
 999                                       " has been read or unknown object")
1000     target.serial_no += 1
1001
1002     if update_serial:
1003       # for node updates, we need to increase the cluster serial too
1004       self._config_data.cluster.serial_no += 1
1005
1006     self._WriteConfig()