Statistics
| Branch: | Tag: | Revision:

root / lib / config.py @ ec0292f1

History | View | Annotate | Download (33.6 kB)

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()