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