Statistics
| Branch: | Tag: | Revision:

root / lib / config.py @ 03d1dba2

History | View | Annotate | Download (30.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
    """Stub verify function.
218
    """
219
    result = []
220
    seen_macs = []
221
    ports = {}
222
    data = self._config_data
223
    for instance_name in data.instances:
224
      instance = data.instances[instance_name]
225
      if instance.primary_node not in data.nodes:
226
        result.append("instance '%s' has invalid primary node '%s'" %
227
                      (instance_name, instance.primary_node))
228
      for snode in instance.secondary_nodes:
229
        if snode not in data.nodes:
230
          result.append("instance '%s' has invalid secondary node '%s'" %
231
                        (instance_name, snode))
232
      for idx, nic in enumerate(instance.nics):
233
        if nic.mac in seen_macs:
234
          result.append("instance '%s' has NIC %d mac %s duplicate" %
235
                        (instance_name, idx, nic.mac))
236
        else:
237
          seen_macs.append(nic.mac)
238

    
239
      # gather the drbd ports for duplicate checks
240
      for dsk in instance.disks:
241
        if dsk.dev_type in constants.LDS_DRBD:
242
          tcp_port = dsk.logical_id[2]
243
          if tcp_port not in ports:
244
            ports[tcp_port] = []
245
          ports[tcp_port].append((instance.name, "drbd disk %s" % dsk.iv_name))
246
      # gather network port reservation
247
      net_port = getattr(instance, "network_port", None)
248
      if net_port is not None:
249
        if net_port not in ports:
250
          ports[net_port] = []
251
        ports[net_port].append((instance.name, "network port"))
252

    
253
    # cluster-wide pool of free ports
254
    for free_port in self._config_data.cluster.tcpudp_port_pool:
255
      if free_port not in ports:
256
        ports[free_port] = []
257
      ports[free_port].append(("cluster", "port marked as free"))
258

    
259
    # compute tcp/udp duplicate ports
260
    keys = ports.keys()
261
    keys.sort()
262
    for pnum in keys:
263
      pdata = ports[pnum]
264
      if len(pdata) > 1:
265
        txt = ", ".join(["%s/%s" % val for val in pdata])
266
        result.append("tcp/udp port %s has duplicates: %s" % (pnum, txt))
267

    
268
    # highest used tcp port check
269
    if keys:
270
      if keys[-1] > self._config_data.cluster.highest_used_port:
271
        result.append("Highest used port mismatch, saved %s, computed %s" %
272
                      (self._config_data.cluster.highest_used_port,
273
                       keys[-1]))
274

    
275
    return result
276

    
277
  def _UnlockedSetDiskID(self, disk, node_name):
278
    """Convert the unique ID to the ID needed on the target nodes.
279

280
    This is used only for drbd, which needs ip/port configuration.
281

282
    The routine descends down and updates its children also, because
283
    this helps when the only the top device is passed to the remote
284
    node.
285

286
    This function is for internal use, when the config lock is already held.
287

288
    """
289
    if disk.children:
290
      for child in disk.children:
291
        self._UnlockedSetDiskID(child, node_name)
292

    
293
    if disk.logical_id is None and disk.physical_id is not None:
294
      return
295
    if disk.dev_type == constants.LD_DRBD8:
296
      pnode, snode, port, pminor, sminor, secret = disk.logical_id
297
      if node_name not in (pnode, snode):
298
        raise errors.ConfigurationError("DRBD device not knowing node %s" %
299
                                        node_name)
300
      pnode_info = self._UnlockedGetNodeInfo(pnode)
301
      snode_info = self._UnlockedGetNodeInfo(snode)
302
      if pnode_info is None or snode_info is None:
303
        raise errors.ConfigurationError("Can't find primary or secondary node"
304
                                        " for %s" % str(disk))
305
      p_data = (pnode_info.secondary_ip, port)
306
      s_data = (snode_info.secondary_ip, port)
307
      if pnode == node_name:
308
        disk.physical_id = p_data + s_data + (pminor, secret)
309
      else: # it must be secondary, we tested above
310
        disk.physical_id = s_data + p_data + (sminor, secret)
311
    else:
312
      disk.physical_id = disk.logical_id
313
    return
314

    
315
  @locking.ssynchronized(_config_lock)
316
  def SetDiskID(self, disk, node_name):
317
    """Convert the unique ID to the ID needed on the target nodes.
318

319
    This is used only for drbd, which needs ip/port configuration.
320

321
    The routine descends down and updates its children also, because
322
    this helps when the only the top device is passed to the remote
323
    node.
324

325
    """
326
    return self._UnlockedSetDiskID(disk, node_name)
327

    
328
  @locking.ssynchronized(_config_lock)
329
  def AddTcpUdpPort(self, port):
330
    """Adds a new port to the available port pool.
331

332
    """
333
    if not isinstance(port, int):
334
      raise errors.ProgrammerError("Invalid type passed for port")
335

    
336
    self._config_data.cluster.tcpudp_port_pool.add(port)
337
    self._WriteConfig()
338

    
339
  @locking.ssynchronized(_config_lock, shared=1)
340
  def GetPortList(self):
341
    """Returns a copy of the current port list.
342

343
    """
344
    return self._config_data.cluster.tcpudp_port_pool.copy()
345

    
346
  @locking.ssynchronized(_config_lock)
347
  def AllocatePort(self):
348
    """Allocate a port.
349

350
    The port will be taken from the available port pool or from the
351
    default port range (and in this case we increase
352
    highest_used_port).
353

354
    """
355
    # If there are TCP/IP ports configured, we use them first.
356
    if self._config_data.cluster.tcpudp_port_pool:
357
      port = self._config_data.cluster.tcpudp_port_pool.pop()
358
    else:
359
      port = self._config_data.cluster.highest_used_port + 1
360
      if port >= constants.LAST_DRBD_PORT:
361
        raise errors.ConfigurationError("The highest used port is greater"
362
                                        " than %s. Aborting." %
363
                                        constants.LAST_DRBD_PORT)
364
      self._config_data.cluster.highest_used_port = port
365

    
366
    self._WriteConfig()
367
    return port
368

    
369
  def _ComputeDRBDMap(self, instance):
370
    """Compute the used DRBD minor/nodes.
371

372
    Return: dictionary of node_name: dict of minor: instance_name. The
373
    returned dict will have all the nodes in it (even if with an empty
374
    list).
375

376
    """
377
    def _AppendUsedPorts(instance_name, disk, used):
378
      if disk.dev_type == constants.LD_DRBD8 and len(disk.logical_id) >= 5:
379
        nodeA, nodeB, dummy, minorA, minorB = disk.logical_id[:5]
380
        for node, port in ((nodeA, minorA), (nodeB, minorB)):
381
          assert node in used, "Instance node not found in node list"
382
          if port in used[node]:
383
            raise errors.ProgrammerError("DRBD minor already used:"
384
                                         " %s/%s, %s/%s" %
385
                                         (node, port, instance_name,
386
                                          used[node][port]))
387

    
388
          used[node][port] = instance_name
389
      if disk.children:
390
        for child in disk.children:
391
          _AppendUsedPorts(instance_name, child, used)
392

    
393
    my_dict = dict((node, {}) for node in self._config_data.nodes)
394
    for (node, minor), instance in self._temporary_drbds.iteritems():
395
      my_dict[node][minor] = instance
396
    for instance in self._config_data.instances.itervalues():
397
      for disk in instance.disks:
398
        _AppendUsedPorts(instance.name, disk, my_dict)
399
    return my_dict
400

    
401
  @locking.ssynchronized(_config_lock)
402
  def AllocateDRBDMinor(self, nodes, instance):
403
    """Allocate a drbd minor.
404

405
    The free minor will be automatically computed from the existing
406
    devices. A node can be given multiple times in order to allocate
407
    multiple minors. The result is the list of minors, in the same
408
    order as the passed nodes.
409

410
    """
411
    d_map = self._ComputeDRBDMap(instance)
412
    result = []
413
    for nname in nodes:
414
      ndata = d_map[nname]
415
      if not ndata:
416
        # no minors used, we can start at 0
417
        result.append(0)
418
        ndata[0] = instance
419
        self._temporary_drbds[(nname, 0)] = instance
420
        continue
421
      keys = ndata.keys()
422
      keys.sort()
423
      ffree = utils.FirstFree(keys)
424
      if ffree is None:
425
        # return the next minor
426
        # TODO: implement high-limit check
427
        minor = keys[-1] + 1
428
      else:
429
        minor = ffree
430
      result.append(minor)
431
      ndata[minor] = instance
432
      assert (nname, minor) not in self._temporary_drbds, \
433
             "Attempt to reuse reserved DRBD minor"
434
      self._temporary_drbds[(nname, minor)] = instance
435
    logging.debug("Request to allocate drbd minors, input: %s, returning %s",
436
                  nodes, result)
437
    return result
438

    
439
  @locking.ssynchronized(_config_lock)
440
  def ReleaseDRBDMinors(self, instance):
441
    """Release temporary drbd minors allocated for a given instance.
442

443
    This should be called on both the error paths and on the success
444
    paths (after the instance has been added or updated).
445

446
    @type instance: string
447
    @param instance: the instance for which temporary minors should be
448
                     released
449

450
    """
451
    for key, name in self._temporary_drbds.items():
452
      if name == instance:
453
        del self._temporary_drbds[key]
454

    
455
  @locking.ssynchronized(_config_lock, shared=1)
456
  def GetConfigVersion(self):
457
    """Get the configuration version.
458

459
    @return: Config version
460

461
    """
462
    return self._config_data.version
463

    
464
  @locking.ssynchronized(_config_lock, shared=1)
465
  def GetClusterName(self):
466
    """Get cluster name.
467

468
    @return: Cluster name
469

470
    """
471
    return self._config_data.cluster.cluster_name
472

    
473
  @locking.ssynchronized(_config_lock, shared=1)
474
  def GetMasterNode(self):
475
    """Get the hostname of the master node for this cluster.
476

477
    @return: Master hostname
478

479
    """
480
    return self._config_data.cluster.master_node
481

    
482
  @locking.ssynchronized(_config_lock, shared=1)
483
  def GetMasterIP(self):
484
    """Get the IP of the master node for this cluster.
485

486
    @return: Master IP
487

488
    """
489
    return self._config_data.cluster.master_ip
490

    
491
  @locking.ssynchronized(_config_lock, shared=1)
492
  def GetMasterNetdev(self):
493
    """Get the master network device for this cluster.
494

495
    """
496
    return self._config_data.cluster.master_netdev
497

    
498
  @locking.ssynchronized(_config_lock, shared=1)
499
  def GetFileStorageDir(self):
500
    """Get the file storage dir for this cluster.
501

502
    """
503
    return self._config_data.cluster.file_storage_dir
504

    
505
  @locking.ssynchronized(_config_lock, shared=1)
506
  def GetHypervisorType(self):
507
    """Get the hypervisor type for this cluster.
508

509
    """
510
    return self._config_data.cluster.default_hypervisor
511

    
512
  @locking.ssynchronized(_config_lock, shared=1)
513
  def GetHostKey(self):
514
    """Return the rsa hostkey from the config.
515

516
    Args: None
517

518
    Returns: rsa hostkey
519
    """
520
    return self._config_data.cluster.rsahostkeypub
521

    
522
  @locking.ssynchronized(_config_lock)
523
  def AddInstance(self, instance):
524
    """Add an instance to the config.
525

526
    This should be used after creating a new instance.
527

528
    Args:
529
      instance: the instance object
530
    """
531
    if not isinstance(instance, objects.Instance):
532
      raise errors.ProgrammerError("Invalid type passed to AddInstance")
533

    
534
    if instance.disk_template != constants.DT_DISKLESS:
535
      all_lvs = instance.MapLVsByNode()
536
      logging.info("Instance '%s' DISK_LAYOUT: %s", instance.name, all_lvs)
537

    
538
    instance.serial_no = 1
539
    self._config_data.instances[instance.name] = instance
540
    self._config_data.cluster.serial_no += 1
541
    self._WriteConfig()
542

    
543
  def _SetInstanceStatus(self, instance_name, status):
544
    """Set the instance's status to a given value.
545

546
    """
547
    if status not in ("up", "down"):
548
      raise errors.ProgrammerError("Invalid status '%s' passed to"
549
                                   " ConfigWriter._SetInstanceStatus()" %
550
                                   status)
551

    
552
    if instance_name not in self._config_data.instances:
553
      raise errors.ConfigurationError("Unknown instance '%s'" %
554
                                      instance_name)
555
    instance = self._config_data.instances[instance_name]
556
    if instance.status != status:
557
      instance.status = status
558
      instance.serial_no += 1
559
      self._WriteConfig()
560

    
561
  @locking.ssynchronized(_config_lock)
562
  def MarkInstanceUp(self, instance_name):
563
    """Mark the instance status to up in the config.
564

565
    """
566
    self._SetInstanceStatus(instance_name, "up")
567

    
568
  @locking.ssynchronized(_config_lock)
569
  def RemoveInstance(self, instance_name):
570
    """Remove the instance from the configuration.
571

572
    """
573
    if instance_name not in self._config_data.instances:
574
      raise errors.ConfigurationError("Unknown instance '%s'" % instance_name)
575
    del self._config_data.instances[instance_name]
576
    self._config_data.cluster.serial_no += 1
577
    self._WriteConfig()
578

    
579
  @locking.ssynchronized(_config_lock)
580
  def RenameInstance(self, old_name, new_name):
581
    """Rename an instance.
582

583
    This needs to be done in ConfigWriter and not by RemoveInstance
584
    combined with AddInstance as only we can guarantee an atomic
585
    rename.
586

587
    """
588
    if old_name not in self._config_data.instances:
589
      raise errors.ConfigurationError("Unknown instance '%s'" % old_name)
590
    inst = self._config_data.instances[old_name]
591
    del self._config_data.instances[old_name]
592
    inst.name = new_name
593

    
594
    for disk in inst.disks:
595
      if disk.dev_type == constants.LD_FILE:
596
        # rename the file paths in logical and physical id
597
        file_storage_dir = os.path.dirname(os.path.dirname(disk.logical_id[1]))
598
        disk.physical_id = disk.logical_id = (disk.logical_id[0],
599
                                              os.path.join(file_storage_dir,
600
                                                           inst.name,
601
                                                           disk.iv_name))
602

    
603
    self._config_data.instances[inst.name] = inst
604
    self._config_data.cluster.serial_no += 1
605
    self._WriteConfig()
606

    
607
  @locking.ssynchronized(_config_lock)
608
  def MarkInstanceDown(self, instance_name):
609
    """Mark the status of an instance to down in the configuration.
610

611
    """
612
    self._SetInstanceStatus(instance_name, "down")
613

    
614
  def _UnlockedGetInstanceList(self):
615
    """Get the list of instances.
616

617
    This function is for internal use, when the config lock is already held.
618

619
    """
620
    return self._config_data.instances.keys()
621

    
622
  @locking.ssynchronized(_config_lock, shared=1)
623
  def GetInstanceList(self):
624
    """Get the list of instances.
625

626
    Returns:
627
      array of instances, ex. ['instance2.example.com','instance1.example.com']
628
      these contains all the instances, also the ones in Admin_down state
629

630
    """
631
    return self._UnlockedGetInstanceList()
632

    
633
  @locking.ssynchronized(_config_lock, shared=1)
634
  def ExpandInstanceName(self, short_name):
635
    """Attempt to expand an incomplete instance name.
636

637
    """
638
    return utils.MatchNameComponent(short_name,
639
                                    self._config_data.instances.keys())
640

    
641
  def _UnlockedGetInstanceInfo(self, instance_name):
642
    """Returns informations about an instance.
643

644
    This function is for internal use, when the config lock is already held.
645

646
    """
647
    if instance_name not in self._config_data.instances:
648
      return None
649

    
650
    return self._config_data.instances[instance_name]
651

    
652
  @locking.ssynchronized(_config_lock, shared=1)
653
  def GetInstanceInfo(self, instance_name):
654
    """Returns informations about an instance.
655

656
    It takes the information from the configuration file. Other informations of
657
    an instance are taken from the live systems.
658

659
    Args:
660
      instance: name of the instance, ex instance1.example.com
661

662
    Returns:
663
      the instance object
664

665
    """
666
    return self._UnlockedGetInstanceInfo(instance_name)
667

    
668
  @locking.ssynchronized(_config_lock, shared=1)
669
  def GetAllInstancesInfo(self):
670
    """Get the configuration of all instances.
671

672
    @rtype: dict
673
    @returns: dict of (instance, instance_info), where instance_info is what
674
              would GetInstanceInfo return for the node
675

676
    """
677
    my_dict = dict([(instance, self._UnlockedGetInstanceInfo(instance))
678
                    for instance in self._UnlockedGetInstanceList()])
679
    return my_dict
680

    
681
  @locking.ssynchronized(_config_lock)
682
  def AddNode(self, node):
683
    """Add a node to the configuration.
684

685
    Args:
686
      node: an object.Node instance
687

688
    """
689
    logging.info("Adding node %s to configuration" % node.name)
690

    
691
    node.serial_no = 1
692
    self._config_data.nodes[node.name] = node
693
    self._config_data.cluster.serial_no += 1
694
    self._WriteConfig()
695

    
696
  @locking.ssynchronized(_config_lock)
697
  def RemoveNode(self, node_name):
698
    """Remove a node from the configuration.
699

700
    """
701
    logging.info("Removing node %s from configuration" % node_name)
702

    
703
    if node_name not in self._config_data.nodes:
704
      raise errors.ConfigurationError("Unknown node '%s'" % node_name)
705

    
706
    del self._config_data.nodes[node_name]
707
    self._config_data.cluster.serial_no += 1
708
    self._WriteConfig()
709

    
710
  @locking.ssynchronized(_config_lock, shared=1)
711
  def ExpandNodeName(self, short_name):
712
    """Attempt to expand an incomplete instance name.
713

714
    """
715
    return utils.MatchNameComponent(short_name,
716
                                    self._config_data.nodes.keys())
717

    
718
  def _UnlockedGetNodeInfo(self, node_name):
719
    """Get the configuration of a node, as stored in the config.
720

721
    This function is for internal use, when the config lock is already held.
722

723
    Args: node: nodename (tuple) of the node
724

725
    Returns: the node object
726

727
    """
728
    if node_name not in self._config_data.nodes:
729
      return None
730

    
731
    return self._config_data.nodes[node_name]
732

    
733

    
734
  @locking.ssynchronized(_config_lock, shared=1)
735
  def GetNodeInfo(self, node_name):
736
    """Get the configuration of a node, as stored in the config.
737

738
    Args: node: nodename (tuple) of the node
739

740
    Returns: the node object
741

742
    """
743
    return self._UnlockedGetNodeInfo(node_name)
744

    
745
  def _UnlockedGetNodeList(self):
746
    """Return the list of nodes which are in the configuration.
747

748
    This function is for internal use, when the config lock is already held.
749

750
    """
751
    return self._config_data.nodes.keys()
752

    
753

    
754
  @locking.ssynchronized(_config_lock, shared=1)
755
  def GetNodeList(self):
756
    """Return the list of nodes which are in the configuration.
757

758
    """
759
    return self._UnlockedGetNodeList()
760

    
761
  @locking.ssynchronized(_config_lock, shared=1)
762
  def GetAllNodesInfo(self):
763
    """Get the configuration of all nodes.
764

765
    @rtype: dict
766
    @returns: dict of (node, node_info), where node_info is what
767
              would GetNodeInfo return for the node
768

769
    """
770
    my_dict = dict([(node, self._UnlockedGetNodeInfo(node))
771
                    for node in self._UnlockedGetNodeList()])
772
    return my_dict
773

    
774
  def _BumpSerialNo(self):
775
    """Bump up the serial number of the config.
776

777
    """
778
    self._config_data.serial_no += 1
779

    
780
  def _OpenConfig(self):
781
    """Read the config data from disk.
782

783
    In case we already have configuration data and the config file has
784
    the same mtime as when we read it, we skip the parsing of the
785
    file, since de-serialisation could be slow.
786

787
    """
788
    f = open(self._cfg_file, 'r')
789
    try:
790
      try:
791
        data = objects.ConfigData.FromDict(serializer.Load(f.read()))
792
      except Exception, err:
793
        raise errors.ConfigurationError(err)
794
    finally:
795
      f.close()
796

    
797
    # Make sure the configuration has the right version
798
    _ValidateConfig(data)
799

    
800
    if (not hasattr(data, 'cluster') or
801
        not hasattr(data.cluster, 'rsahostkeypub')):
802
      raise errors.ConfigurationError("Incomplete configuration"
803
                                      " (missing cluster.rsahostkeypub)")
804
    self._config_data = data
805
    # init the last serial as -1 so that the next write will cause
806
    # ssconf update
807
    self._last_cluster_serial = -1
808

    
809
  def _DistributeConfig(self):
810
    """Distribute the configuration to the other nodes.
811

812
    Currently, this only copies the configuration file. In the future,
813
    it could be used to encapsulate the 2/3-phase update mechanism.
814

815
    """
816
    if self._offline:
817
      return True
818
    bad = False
819
    nodelist = self._UnlockedGetNodeList()
820
    myhostname = self._my_hostname
821

    
822
    try:
823
      nodelist.remove(myhostname)
824
    except ValueError:
825
      pass
826
    # we can skip checking whether _UnlockedGetNodeInfo returns None
827
    # since the node list comes from _UnlocketGetNodeList, and we are
828
    # called with the lock held, so no modifications should take place
829
    # in between
830
    address_list = [self._UnlockedGetNodeInfo(name).primary_ip
831
                    for name in nodelist]
832

    
833
    result = rpc.RpcRunner.call_upload_file(nodelist, self._cfg_file,
834
                                            address_list=address_list)
835
    for node in nodelist:
836
      if not result[node]:
837
        logging.error("copy of file %s to node %s failed",
838
                      self._cfg_file, node)
839
        bad = True
840
    return not bad
841

    
842
  def _WriteConfig(self, destination=None):
843
    """Write the configuration data to persistent storage.
844

845
    """
846
    if destination is None:
847
      destination = self._cfg_file
848
    self._BumpSerialNo()
849
    txt = serializer.Dump(self._config_data.ToDict())
850
    dir_name, file_name = os.path.split(destination)
851
    fd, name = tempfile.mkstemp('.newconfig', file_name, dir_name)
852
    f = os.fdopen(fd, 'w')
853
    try:
854
      f.write(txt)
855
      os.fsync(f.fileno())
856
    finally:
857
      f.close()
858
    # we don't need to do os.close(fd) as f.close() did it
859
    os.rename(name, destination)
860
    self.write_count += 1
861

    
862
    # and redistribute the config file
863
    self._DistributeConfig()
864

    
865
    # Write ssconf files on all nodes (including locally)
866
    if self._last_cluster_serial < self._config_data.cluster.serial_no:
867
      if not self._offline:
868
        rpc.RpcRunner.call_write_ssconf_files(self._UnlockedGetNodeList(),
869
                                              self._UnlockedGetSsconfValues())
870
      self._last_cluster_serial = self._config_data.cluster.serial_no
871

    
872
  def _UnlockedGetSsconfValues(self):
873
    return {
874
      "cluster_name": self._config_data.cluster.cluster_name,
875
      "master_ip": self._config_data.cluster.master_ip,
876
      "master_netdev": self._config_data.cluster.master_netdev,
877
      "master_node": self._config_data.cluster.master_node,
878
      }
879

    
880
  @locking.ssynchronized(_config_lock)
881
  def InitConfig(self, version, cluster_config, master_node_config):
882
    """Create the initial cluster configuration.
883

884
    It will contain the current node, which will also be the master
885
    node, and no instances.
886

887
    @type version: int
888
    @param version: Configuration version
889
    @type cluster_config: objects.Cluster
890
    @param cluster_config: Cluster configuration
891
    @type master_node_config: objects.Node
892
    @param master_node_config: Master node configuration
893

894
    """
895
    nodes = {
896
      master_node_config.name: master_node_config,
897
      }
898

    
899
    self._config_data = objects.ConfigData(version=version,
900
                                           cluster=cluster_config,
901
                                           nodes=nodes,
902
                                           instances={},
903
                                           serial_no=1)
904
    self._WriteConfig()
905

    
906
  @locking.ssynchronized(_config_lock, shared=1)
907
  def GetVGName(self):
908
    """Return the volume group name.
909

910
    """
911
    return self._config_data.cluster.volume_group_name
912

    
913
  @locking.ssynchronized(_config_lock)
914
  def SetVGName(self, vg_name):
915
    """Set the volume group name.
916

917
    """
918
    self._config_data.cluster.volume_group_name = vg_name
919
    self._config_data.cluster.serial_no += 1
920
    self._WriteConfig()
921

    
922
  @locking.ssynchronized(_config_lock, shared=1)
923
  def GetDefBridge(self):
924
    """Return the default bridge.
925

926
    """
927
    return self._config_data.cluster.default_bridge
928

    
929
  @locking.ssynchronized(_config_lock, shared=1)
930
  def GetMACPrefix(self):
931
    """Return the mac prefix.
932

933
    """
934
    return self._config_data.cluster.mac_prefix
935

    
936
  @locking.ssynchronized(_config_lock, shared=1)
937
  def GetClusterInfo(self):
938
    """Returns informations about the cluster
939

940
    Returns:
941
      the cluster object
942

943
    """
944
    return self._config_data.cluster
945

    
946
  @locking.ssynchronized(_config_lock)
947
  def Update(self, target):
948
    """Notify function to be called after updates.
949

950
    This function must be called when an object (as returned by
951
    GetInstanceInfo, GetNodeInfo, GetCluster) has been updated and the
952
    caller wants the modifications saved to the backing store. Note
953
    that all modified objects will be saved, but the target argument
954
    is the one the caller wants to ensure that it's saved.
955

956
    """
957
    if self._config_data is None:
958
      raise errors.ProgrammerError("Configuration file not read,"
959
                                   " cannot save.")
960
    if isinstance(target, objects.Cluster):
961
      test = target == self._config_data.cluster
962
    elif isinstance(target, objects.Node):
963
      test = target in self._config_data.nodes.values()
964
    elif isinstance(target, objects.Instance):
965
      test = target in self._config_data.instances.values()
966
    else:
967
      raise errors.ProgrammerError("Invalid object type (%s) passed to"
968
                                   " ConfigWriter.Update" % type(target))
969
    if not test:
970
      raise errors.ConfigurationError("Configuration updated since object"
971
                                      " has been read or unknown object")
972
    target.serial_no += 1
973

    
974
    self._WriteConfig()