Statistics
| Branch: | Tag: | Revision:

root / lib / config.py @ a8efbb40

History | View | Annotate | Download (31.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
    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()