Statistics
| Branch: | Tag: | Revision:

root / lib / config.py @ 79b26a7a

History | View | Annotate | Download (37.7 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
  """Verifies that a configuration objects looks valid.
53

54
  This only verifies the version of the configuration.
55

56
  @raise errors.ConfigurationError: if the version differs from what
57
      we expect
58

59
  """
60
  if data.version != constants.CONFIG_VERSION:
61
    raise errors.ConfigurationError("Cluster configuration version"
62
                                    " mismatch, got %s instead of %s" %
63
                                    (data.version,
64
                                     constants.CONFIG_VERSION))
65

    
66

    
67
class ConfigWriter:
68
  """The interface to the cluster configuration.
69

70
  """
71
  def __init__(self, cfg_file=None, offline=False):
72
    self.write_count = 0
73
    self._lock = _config_lock
74
    self._config_data = None
75
    self._offline = offline
76
    if cfg_file is None:
77
      self._cfg_file = constants.CLUSTER_CONF_FILE
78
    else:
79
      self._cfg_file = cfg_file
80
    self._temporary_ids = set()
81
    self._temporary_drbds = {}
82
    # Note: in order to prevent errors when resolving our name in
83
    # _DistributeConfig, we compute it here once and reuse it; it's
84
    # better to raise an error before starting to modify the config
85
    # file than after it was modified
86
    self._my_hostname = utils.HostInfo().name
87
    self._last_cluster_serial = -1
88
    self._OpenConfig()
89

    
90
  # this method needs to be static, so that we can call it on the class
91
  @staticmethod
92
  def IsCluster():
93
    """Check if the cluster is configured.
94

95
    """
96
    return os.path.exists(constants.CLUSTER_CONF_FILE)
97

    
98
  @locking.ssynchronized(_config_lock, shared=1)
99
  def GenerateMAC(self):
100
    """Generate a MAC for an instance.
101

102
    This should check the current instances for duplicates.
103

104
    """
105
    prefix = self._config_data.cluster.mac_prefix
106
    all_macs = self._AllMACs()
107
    retries = 64
108
    while retries > 0:
109
      byte1 = random.randrange(0, 256)
110
      byte2 = random.randrange(0, 256)
111
      byte3 = random.randrange(0, 256)
112
      mac = "%s:%02x:%02x:%02x" % (prefix, byte1, byte2, byte3)
113
      if mac not in all_macs:
114
        break
115
      retries -= 1
116
    else:
117
      raise errors.ConfigurationError("Can't generate unique MAC")
118
    return mac
119

    
120
  @locking.ssynchronized(_config_lock, shared=1)
121
  def IsMacInUse(self, mac):
122
    """Predicate: check if the specified MAC is in use in the Ganeti cluster.
123

124
    This only checks instances managed by this cluster, it does not
125
    check for potential collisions elsewhere.
126

127
    """
128
    all_macs = self._AllMACs()
129
    return mac in all_macs
130

    
131
  @locking.ssynchronized(_config_lock, shared=1)
132
  def GenerateDRBDSecret(self):
133
    """Generate a DRBD secret.
134

135
    This checks the current disks for duplicates.
136

137
    """
138
    all_secrets = self._AllDRBDSecrets()
139
    retries = 64
140
    while retries > 0:
141
      secret = utils.GenerateSecret()
142
      if secret not in all_secrets:
143
        break
144
      retries -= 1
145
    else:
146
      raise errors.ConfigurationError("Can't generate unique DRBD secret")
147
    return secret
148

    
149
  def _ComputeAllLVs(self):
150
    """Compute the list of all LVs.
151

152
    """
153
    lvnames = set()
154
    for instance in self._config_data.instances.values():
155
      node_data = instance.MapLVsByNode()
156
      for lv_list in node_data.values():
157
        lvnames.update(lv_list)
158
    return lvnames
159

    
160
  @locking.ssynchronized(_config_lock, shared=1)
161
  def GenerateUniqueID(self, exceptions=None):
162
    """Generate an unique disk name.
163

164
    This checks the current node, instances and disk names for
165
    duplicates.
166

167
    @param exceptions: a list with some other names which should be checked
168
        for uniqueness (used for example when you want to get
169
        more than one id at one time without adding each one in
170
        turn to the config file)
171

172
    @rtype: string
173
    @return: the unique id
174

175
    """
176
    existing = set()
177
    existing.update(self._temporary_ids)
178
    existing.update(self._ComputeAllLVs())
179
    existing.update(self._config_data.instances.keys())
180
    existing.update(self._config_data.nodes.keys())
181
    if exceptions is not None:
182
      existing.update(exceptions)
183
    retries = 64
184
    while retries > 0:
185
      unique_id = utils.NewUUID()
186
      if unique_id not in existing and unique_id is not None:
187
        break
188
    else:
189
      raise errors.ConfigurationError("Not able generate an unique ID"
190
                                      " (last tried ID: %s" % unique_id)
191
    self._temporary_ids.add(unique_id)
192
    return unique_id
193

    
194
  def _AllMACs(self):
195
    """Return all MACs present in the config.
196

197
    @rtype: list
198
    @return: the list of all MACs
199

200
    """
201
    result = []
202
    for instance in self._config_data.instances.values():
203
      for nic in instance.nics:
204
        result.append(nic.mac)
205

    
206
    return result
207

    
208
  def _AllDRBDSecrets(self):
209
    """Return all DRBD secrets present in the config.
210

211
    @rtype: list
212
    @return: the list of all DRBD secrets
213

214
    """
215
    def helper(disk, result):
216
      """Recursively gather secrets from this disk."""
217
      if disk.dev_type == constants.DT_DRBD8:
218
        result.append(disk.logical_id[5])
219
      if disk.children:
220
        for child in disk.children:
221
          helper(child, result)
222

    
223
    result = []
224
    for instance in self._config_data.instances.values():
225
      for disk in instance.disks:
226
        helper(disk, result)
227

    
228
    return result
229

    
230
  def _UnlockedVerifyConfig(self):
231
    """Verify function.
232

233
    @rtype: list
234
    @return: a list of error messages; a non-empty list signifies
235
        configuration errors
236

237
    """
238
    result = []
239
    seen_macs = []
240
    ports = {}
241
    data = self._config_data
242
    for instance_name in data.instances:
243
      instance = data.instances[instance_name]
244
      if instance.primary_node not in data.nodes:
245
        result.append("instance '%s' has invalid primary node '%s'" %
246
                      (instance_name, instance.primary_node))
247
      for snode in instance.secondary_nodes:
248
        if snode not in data.nodes:
249
          result.append("instance '%s' has invalid secondary node '%s'" %
250
                        (instance_name, snode))
251
      for idx, nic in enumerate(instance.nics):
252
        if nic.mac in seen_macs:
253
          result.append("instance '%s' has NIC %d mac %s duplicate" %
254
                        (instance_name, idx, nic.mac))
255
        else:
256
          seen_macs.append(nic.mac)
257

    
258
      # gather the drbd ports for duplicate checks
259
      for dsk in instance.disks:
260
        if dsk.dev_type in constants.LDS_DRBD:
261
          tcp_port = dsk.logical_id[2]
262
          if tcp_port not in ports:
263
            ports[tcp_port] = []
264
          ports[tcp_port].append((instance.name, "drbd disk %s" % dsk.iv_name))
265
      # gather network port reservation
266
      net_port = getattr(instance, "network_port", None)
267
      if net_port is not None:
268
        if net_port not in ports:
269
          ports[net_port] = []
270
        ports[net_port].append((instance.name, "network port"))
271

    
272
    # cluster-wide pool of free ports
273
    for free_port in data.cluster.tcpudp_port_pool:
274
      if free_port not in ports:
275
        ports[free_port] = []
276
      ports[free_port].append(("cluster", "port marked as free"))
277

    
278
    # compute tcp/udp duplicate ports
279
    keys = ports.keys()
280
    keys.sort()
281
    for pnum in keys:
282
      pdata = ports[pnum]
283
      if len(pdata) > 1:
284
        txt = ", ".join(["%s/%s" % val for val in pdata])
285
        result.append("tcp/udp port %s has duplicates: %s" % (pnum, txt))
286

    
287
    # highest used tcp port check
288
    if keys:
289
      if keys[-1] > data.cluster.highest_used_port:
290
        result.append("Highest used port mismatch, saved %s, computed %s" %
291
                      (data.cluster.highest_used_port, keys[-1]))
292

    
293
    if not data.nodes[data.cluster.master_node].master_candidate:
294
      result.append("Master node is not a master candidate")
295

    
296
    # master candidate checks
297
    mc_now, mc_max = self._UnlockedGetMasterCandidateStats()
298
    if mc_now < mc_max:
299
      result.append("Not enough master candidates: actual %d, target %d" %
300
                    (mc_now, mc_max))
301

    
302
    # drbd minors check
303
    d_map, duplicates = self._UnlockedComputeDRBDMap()
304
    for node, minor, instance_a, instance_b in duplicates:
305
      result.append("DRBD minor %d on node %s is assigned twice to instances"
306
                    " %s and %s" % (minor, node, instance_a, instance_b))
307

    
308
    return result
309

    
310
  @locking.ssynchronized(_config_lock, shared=1)
311
  def VerifyConfig(self):
312
    """Verify function.
313

314
    This is just a wrapper over L{_UnlockedVerifyConfig}.
315

316
    @rtype: list
317
    @return: a list of error messages; a non-empty list signifies
318
        configuration errors
319

320
    """
321
    return self._UnlockedVerifyConfig()
322

    
323
  def _UnlockedSetDiskID(self, disk, node_name):
324
    """Convert the unique ID to the ID needed on the target nodes.
325

326
    This is used only for drbd, which needs ip/port configuration.
327

328
    The routine descends down and updates its children also, because
329
    this helps when the only the top device is passed to the remote
330
    node.
331

332
    This function is for internal use, when the config lock is already held.
333

334
    """
335
    if disk.children:
336
      for child in disk.children:
337
        self._UnlockedSetDiskID(child, node_name)
338

    
339
    if disk.logical_id is None and disk.physical_id is not None:
340
      return
341
    if disk.dev_type == constants.LD_DRBD8:
342
      pnode, snode, port, pminor, sminor, secret = disk.logical_id
343
      if node_name not in (pnode, snode):
344
        raise errors.ConfigurationError("DRBD device not knowing node %s" %
345
                                        node_name)
346
      pnode_info = self._UnlockedGetNodeInfo(pnode)
347
      snode_info = self._UnlockedGetNodeInfo(snode)
348
      if pnode_info is None or snode_info is None:
349
        raise errors.ConfigurationError("Can't find primary or secondary node"
350
                                        " for %s" % str(disk))
351
      p_data = (pnode_info.secondary_ip, port)
352
      s_data = (snode_info.secondary_ip, port)
353
      if pnode == node_name:
354
        disk.physical_id = p_data + s_data + (pminor, secret)
355
      else: # it must be secondary, we tested above
356
        disk.physical_id = s_data + p_data + (sminor, secret)
357
    else:
358
      disk.physical_id = disk.logical_id
359
    return
360

    
361
  @locking.ssynchronized(_config_lock)
362
  def SetDiskID(self, disk, node_name):
363
    """Convert the unique ID to the ID needed on the target nodes.
364

365
    This is used only for drbd, which needs ip/port configuration.
366

367
    The routine descends down and updates its children also, because
368
    this helps when the only the top device is passed to the remote
369
    node.
370

371
    """
372
    return self._UnlockedSetDiskID(disk, node_name)
373

    
374
  @locking.ssynchronized(_config_lock)
375
  def AddTcpUdpPort(self, port):
376
    """Adds a new port to the available port pool.
377

378
    """
379
    if not isinstance(port, int):
380
      raise errors.ProgrammerError("Invalid type passed for port")
381

    
382
    self._config_data.cluster.tcpudp_port_pool.add(port)
383
    self._WriteConfig()
384

    
385
  @locking.ssynchronized(_config_lock, shared=1)
386
  def GetPortList(self):
387
    """Returns a copy of the current port list.
388

389
    """
390
    return self._config_data.cluster.tcpudp_port_pool.copy()
391

    
392
  @locking.ssynchronized(_config_lock)
393
  def AllocatePort(self):
394
    """Allocate a port.
395

396
    The port will be taken from the available port pool or from the
397
    default port range (and in this case we increase
398
    highest_used_port).
399

400
    """
401
    # If there are TCP/IP ports configured, we use them first.
402
    if self._config_data.cluster.tcpudp_port_pool:
403
      port = self._config_data.cluster.tcpudp_port_pool.pop()
404
    else:
405
      port = self._config_data.cluster.highest_used_port + 1
406
      if port >= constants.LAST_DRBD_PORT:
407
        raise errors.ConfigurationError("The highest used port is greater"
408
                                        " than %s. Aborting." %
409
                                        constants.LAST_DRBD_PORT)
410
      self._config_data.cluster.highest_used_port = port
411

    
412
    self._WriteConfig()
413
    return port
414

    
415
  def _UnlockedComputeDRBDMap(self):
416
    """Compute the used DRBD minor/nodes.
417

418
    @rtype: (dict, list)
419
    @return: dictionary of node_name: dict of minor: instance_name;
420
        the returned dict will have all the nodes in it (even if with
421
        an empty list), and a list of duplicates; if the duplicates
422
        list is not empty, the configuration is corrupted and its caller
423
        should raise an exception
424

425
    """
426
    def _AppendUsedPorts(instance_name, disk, used):
427
      duplicates = []
428
      if disk.dev_type == constants.LD_DRBD8 and len(disk.logical_id) >= 5:
429
        nodeA, nodeB, dummy, minorA, minorB = disk.logical_id[:5]
430
        for node, port in ((nodeA, minorA), (nodeB, minorB)):
431
          assert node in used, ("Node '%s' of instance '%s' not found"
432
                                " in node list" % (node, instance_name))
433
          if port in used[node]:
434
            duplicates.append((node, port, instance_name, used[node][port]))
435
          else:
436
            used[node][port] = instance_name
437
      if disk.children:
438
        for child in disk.children:
439
          duplicates.extend(_AppendUsedPorts(instance_name, child, used))
440
      return duplicates
441

    
442
    duplicates = []
443
    my_dict = dict((node, {}) for node in self._config_data.nodes)
444
    for instance in self._config_data.instances.itervalues():
445
      for disk in instance.disks:
446
        duplicates.extend(_AppendUsedPorts(instance.name, disk, my_dict))
447
    for (node, minor), instance in self._temporary_drbds.iteritems():
448
      if minor in my_dict[node] and my_dict[node][minor] != instance:
449
        duplicates.append((node, minor, instance, my_dict[node][minor]))
450
      else:
451
        my_dict[node][minor] = instance
452
    return my_dict, duplicates
453

    
454
  @locking.ssynchronized(_config_lock)
455
  def ComputeDRBDMap(self):
456
    """Compute the used DRBD minor/nodes.
457

458
    This is just a wrapper over L{_UnlockedComputeDRBDMap}.
459

460
    @return: dictionary of node_name: dict of minor: instance_name;
461
        the returned dict will have all the nodes in it (even if with
462
        an empty list).
463

464
    """
465
    d_map, duplicates = self._UnlockedComputeDRBDMap()
466
    if duplicates:
467
      raise errors.ConfigurationError("Duplicate DRBD ports detected: %s" %
468
                                      str(duplicates))
469
    return d_map
470

    
471
  @locking.ssynchronized(_config_lock)
472
  def AllocateDRBDMinor(self, nodes, instance):
473
    """Allocate a drbd minor.
474

475
    The free minor will be automatically computed from the existing
476
    devices. A node can be given multiple times in order to allocate
477
    multiple minors. The result is the list of minors, in the same
478
    order as the passed nodes.
479

480
    @type instance: string
481
    @param instance: the instance for which we allocate minors
482

483
    """
484
    assert isinstance(instance, basestring), \
485
           "Invalid argument '%s' passed to AllocateDRBDMinor" % instance
486

    
487
    d_map, duplicates = self._UnlockedComputeDRBDMap()
488
    if duplicates:
489
      raise errors.ConfigurationError("Duplicate DRBD ports detected: %s" %
490
                                      str(duplicates))
491
    result = []
492
    for nname in nodes:
493
      ndata = d_map[nname]
494
      if not ndata:
495
        # no minors used, we can start at 0
496
        result.append(0)
497
        ndata[0] = instance
498
        self._temporary_drbds[(nname, 0)] = instance
499
        continue
500
      keys = ndata.keys()
501
      keys.sort()
502
      ffree = utils.FirstFree(keys)
503
      if ffree is None:
504
        # return the next minor
505
        # TODO: implement high-limit check
506
        minor = keys[-1] + 1
507
      else:
508
        minor = ffree
509
      # double-check minor against current instances
510
      assert minor not in d_map[nname], \
511
             ("Attempt to reuse allocated DRBD minor %d on node %s,"
512
              " already allocated to instance %s" %
513
              (minor, nname, d_map[nname][minor]))
514
      ndata[minor] = instance
515
      # double-check minor against reservation
516
      r_key = (nname, minor)
517
      assert r_key not in self._temporary_drbds, \
518
             ("Attempt to reuse reserved DRBD minor %d on node %s,"
519
              " reserved for instance %s" %
520
              (minor, nname, self._temporary_drbds[r_key]))
521
      self._temporary_drbds[r_key] = instance
522
      result.append(minor)
523
    logging.debug("Request to allocate drbd minors, input: %s, returning %s",
524
                  nodes, result)
525
    return result
526

    
527
  def _UnlockedReleaseDRBDMinors(self, instance):
528
    """Release temporary drbd minors allocated for a given instance.
529

530
    @type instance: string
531
    @param instance: the instance for which temporary minors should be
532
                     released
533

534
    """
535
    assert isinstance(instance, basestring), \
536
           "Invalid argument passed to ReleaseDRBDMinors"
537
    for key, name in self._temporary_drbds.items():
538
      if name == instance:
539
        del self._temporary_drbds[key]
540

    
541
  @locking.ssynchronized(_config_lock)
542
  def ReleaseDRBDMinors(self, instance):
543
    """Release temporary drbd minors allocated for a given instance.
544

545
    This should be called on the error paths, on the success paths
546
    it's automatically called by the ConfigWriter add and update
547
    functions.
548

549
    This function is just a wrapper over L{_UnlockedReleaseDRBDMinors}.
550

551
    @type instance: string
552
    @param instance: the instance for which temporary minors should be
553
                     released
554

555
    """
556
    self._UnlockedReleaseDRBDMinors(instance)
557

    
558
  @locking.ssynchronized(_config_lock, shared=1)
559
  def GetConfigVersion(self):
560
    """Get the configuration version.
561

562
    @return: Config version
563

564
    """
565
    return self._config_data.version
566

    
567
  @locking.ssynchronized(_config_lock, shared=1)
568
  def GetClusterName(self):
569
    """Get cluster name.
570

571
    @return: Cluster name
572

573
    """
574
    return self._config_data.cluster.cluster_name
575

    
576
  @locking.ssynchronized(_config_lock, shared=1)
577
  def GetMasterNode(self):
578
    """Get the hostname of the master node for this cluster.
579

580
    @return: Master hostname
581

582
    """
583
    return self._config_data.cluster.master_node
584

    
585
  @locking.ssynchronized(_config_lock, shared=1)
586
  def GetMasterIP(self):
587
    """Get the IP of the master node for this cluster.
588

589
    @return: Master IP
590

591
    """
592
    return self._config_data.cluster.master_ip
593

    
594
  @locking.ssynchronized(_config_lock, shared=1)
595
  def GetMasterNetdev(self):
596
    """Get the master network device for this cluster.
597

598
    """
599
    return self._config_data.cluster.master_netdev
600

    
601
  @locking.ssynchronized(_config_lock, shared=1)
602
  def GetFileStorageDir(self):
603
    """Get the file storage dir for this cluster.
604

605
    """
606
    return self._config_data.cluster.file_storage_dir
607

    
608
  @locking.ssynchronized(_config_lock, shared=1)
609
  def GetHypervisorType(self):
610
    """Get the hypervisor type for this cluster.
611

612
    """
613
    return self._config_data.cluster.default_hypervisor
614

    
615
  @locking.ssynchronized(_config_lock, shared=1)
616
  def GetHostKey(self):
617
    """Return the rsa hostkey from the config.
618

619
    @rtype: string
620
    @return: the rsa hostkey
621

622
    """
623
    return self._config_data.cluster.rsahostkeypub
624

    
625
  @locking.ssynchronized(_config_lock)
626
  def AddInstance(self, instance):
627
    """Add an instance to the config.
628

629
    This should be used after creating a new instance.
630

631
    @type instance: L{objects.Instance}
632
    @param instance: the instance object
633

634
    """
635
    if not isinstance(instance, objects.Instance):
636
      raise errors.ProgrammerError("Invalid type passed to AddInstance")
637

    
638
    if instance.disk_template != constants.DT_DISKLESS:
639
      all_lvs = instance.MapLVsByNode()
640
      logging.info("Instance '%s' DISK_LAYOUT: %s", instance.name, all_lvs)
641

    
642
    instance.serial_no = 1
643
    self._config_data.instances[instance.name] = instance
644
    self._UnlockedReleaseDRBDMinors(instance.name)
645
    self._WriteConfig()
646

    
647
  def _SetInstanceStatus(self, instance_name, status):
648
    """Set the instance's status to a given value.
649

650
    """
651
    assert isinstance(status, bool), \
652
           "Invalid status '%s' passed to SetInstanceStatus" % (status,)
653

    
654
    if instance_name not in self._config_data.instances:
655
      raise errors.ConfigurationError("Unknown instance '%s'" %
656
                                      instance_name)
657
    instance = self._config_data.instances[instance_name]
658
    if instance.admin_up != status:
659
      instance.admin_up = status
660
      instance.serial_no += 1
661
      self._WriteConfig()
662

    
663
  @locking.ssynchronized(_config_lock)
664
  def MarkInstanceUp(self, instance_name):
665
    """Mark the instance status to up in the config.
666

667
    """
668
    self._SetInstanceStatus(instance_name, True)
669

    
670
  @locking.ssynchronized(_config_lock)
671
  def RemoveInstance(self, instance_name):
672
    """Remove the instance from the configuration.
673

674
    """
675
    if instance_name not in self._config_data.instances:
676
      raise errors.ConfigurationError("Unknown instance '%s'" % instance_name)
677
    del self._config_data.instances[instance_name]
678
    self._WriteConfig()
679

    
680
  @locking.ssynchronized(_config_lock)
681
  def RenameInstance(self, old_name, new_name):
682
    """Rename an instance.
683

684
    This needs to be done in ConfigWriter and not by RemoveInstance
685
    combined with AddInstance as only we can guarantee an atomic
686
    rename.
687

688
    """
689
    if old_name not in self._config_data.instances:
690
      raise errors.ConfigurationError("Unknown instance '%s'" % old_name)
691
    inst = self._config_data.instances[old_name]
692
    del self._config_data.instances[old_name]
693
    inst.name = new_name
694

    
695
    for disk in inst.disks:
696
      if disk.dev_type == constants.LD_FILE:
697
        # rename the file paths in logical and physical id
698
        file_storage_dir = os.path.dirname(os.path.dirname(disk.logical_id[1]))
699
        disk.physical_id = disk.logical_id = (disk.logical_id[0],
700
                                              os.path.join(file_storage_dir,
701
                                                           inst.name,
702
                                                           disk.iv_name))
703

    
704
    self._config_data.instances[inst.name] = inst
705
    self._WriteConfig()
706

    
707
  @locking.ssynchronized(_config_lock)
708
  def MarkInstanceDown(self, instance_name):
709
    """Mark the status of an instance to down in the configuration.
710

711
    """
712
    self._SetInstanceStatus(instance_name, False)
713

    
714
  def _UnlockedGetInstanceList(self):
715
    """Get the list of instances.
716

717
    This function is for internal use, when the config lock is already held.
718

719
    """
720
    return self._config_data.instances.keys()
721

    
722
  @locking.ssynchronized(_config_lock, shared=1)
723
  def GetInstanceList(self):
724
    """Get the list of instances.
725

726
    @return: array of instances, ex. ['instance2.example.com',
727
        'instance1.example.com']
728

729
    """
730
    return self._UnlockedGetInstanceList()
731

    
732
  @locking.ssynchronized(_config_lock, shared=1)
733
  def ExpandInstanceName(self, short_name):
734
    """Attempt to expand an incomplete instance name.
735

736
    """
737
    return utils.MatchNameComponent(short_name,
738
                                    self._config_data.instances.keys())
739

    
740
  def _UnlockedGetInstanceInfo(self, instance_name):
741
    """Returns informations about an instance.
742

743
    This function is for internal use, when the config lock is already held.
744

745
    """
746
    if instance_name not in self._config_data.instances:
747
      return None
748

    
749
    return self._config_data.instances[instance_name]
750

    
751
  @locking.ssynchronized(_config_lock, shared=1)
752
  def GetInstanceInfo(self, instance_name):
753
    """Returns informations about an instance.
754

755
    It takes the information from the configuration file. Other informations of
756
    an instance are taken from the live systems.
757

758
    @param instance_name: name of the instance, e.g.
759
        I{instance1.example.com}
760

761
    @rtype: L{objects.Instance}
762
    @return: the instance object
763

764
    """
765
    return self._UnlockedGetInstanceInfo(instance_name)
766

    
767
  @locking.ssynchronized(_config_lock, shared=1)
768
  def GetAllInstancesInfo(self):
769
    """Get the configuration of all instances.
770

771
    @rtype: dict
772
    @returns: dict of (instance, instance_info), where instance_info is what
773
              would GetInstanceInfo return for the node
774

775
    """
776
    my_dict = dict([(instance, self._UnlockedGetInstanceInfo(instance))
777
                    for instance in self._UnlockedGetInstanceList()])
778
    return my_dict
779

    
780
  @locking.ssynchronized(_config_lock)
781
  def AddNode(self, node):
782
    """Add a node to the configuration.
783

784
    @type node: L{objects.Node}
785
    @param node: a Node instance
786

787
    """
788
    logging.info("Adding node %s to configuration" % node.name)
789

    
790
    node.serial_no = 1
791
    self._config_data.nodes[node.name] = node
792
    self._config_data.cluster.serial_no += 1
793
    self._WriteConfig()
794

    
795
  @locking.ssynchronized(_config_lock)
796
  def RemoveNode(self, node_name):
797
    """Remove a node from the configuration.
798

799
    """
800
    logging.info("Removing node %s from configuration" % node_name)
801

    
802
    if node_name not in self._config_data.nodes:
803
      raise errors.ConfigurationError("Unknown node '%s'" % node_name)
804

    
805
    del self._config_data.nodes[node_name]
806
    self._config_data.cluster.serial_no += 1
807
    self._WriteConfig()
808

    
809
  @locking.ssynchronized(_config_lock, shared=1)
810
  def ExpandNodeName(self, short_name):
811
    """Attempt to expand an incomplete instance name.
812

813
    """
814
    return utils.MatchNameComponent(short_name,
815
                                    self._config_data.nodes.keys())
816

    
817
  def _UnlockedGetNodeInfo(self, node_name):
818
    """Get the configuration of a node, as stored in the config.
819

820
    This function is for internal use, when the config lock is already
821
    held.
822

823
    @param node_name: the node name, e.g. I{node1.example.com}
824

825
    @rtype: L{objects.Node}
826
    @return: the node object
827

828
    """
829
    if node_name not in self._config_data.nodes:
830
      return None
831

    
832
    return self._config_data.nodes[node_name]
833

    
834

    
835
  @locking.ssynchronized(_config_lock, shared=1)
836
  def GetNodeInfo(self, node_name):
837
    """Get the configuration of a node, as stored in the config.
838

839
    This is just a locked wrapper over L{_UnlockedGetNodeInfo}.
840

841
    @param node_name: the node name, e.g. I{node1.example.com}
842

843
    @rtype: L{objects.Node}
844
    @return: the node object
845

846
    """
847
    return self._UnlockedGetNodeInfo(node_name)
848

    
849
  def _UnlockedGetNodeList(self):
850
    """Return the list of nodes which are in the configuration.
851

852
    This function is for internal use, when the config lock is already
853
    held.
854

855
    @rtype: list
856

857
    """
858
    return self._config_data.nodes.keys()
859

    
860

    
861
  @locking.ssynchronized(_config_lock, shared=1)
862
  def GetNodeList(self):
863
    """Return the list of nodes which are in the configuration.
864

865
    """
866
    return self._UnlockedGetNodeList()
867

    
868
  @locking.ssynchronized(_config_lock, shared=1)
869
  def GetOnlineNodeList(self):
870
    """Return the list of nodes which are online.
871

872
    """
873
    all_nodes = [self._UnlockedGetNodeInfo(node)
874
                 for node in self._UnlockedGetNodeList()]
875
    return [node.name for node in all_nodes if not node.offline]
876

    
877
  @locking.ssynchronized(_config_lock, shared=1)
878
  def GetAllNodesInfo(self):
879
    """Get the configuration of all nodes.
880

881
    @rtype: dict
882
    @return: dict of (node, node_info), where node_info is what
883
              would GetNodeInfo return for the node
884

885
    """
886
    my_dict = dict([(node, self._UnlockedGetNodeInfo(node))
887
                    for node in self._UnlockedGetNodeList()])
888
    return my_dict
889

    
890
  def _UnlockedGetMasterCandidateStats(self):
891
    """Get the number of current and maximum desired and possible candidates.
892

893
    @rtype: tuple
894
    @return: tuple of (current, desired and possible)
895

896
    """
897
    mc_now = mc_max = 0
898
    for node in self._config_data.nodes.itervalues():
899
      if not node.offline:
900
        mc_max += 1
901
      if node.master_candidate:
902
        mc_now += 1
903
    mc_max = min(mc_max, self._config_data.cluster.candidate_pool_size)
904
    return (mc_now, mc_max)
905

    
906
  @locking.ssynchronized(_config_lock, shared=1)
907
  def GetMasterCandidateStats(self):
908
    """Get the number of current and maximum possible candidates.
909

910
    This is just a wrapper over L{_UnlockedGetMasterCandidateStats}.
911

912
    @rtype: tuple
913
    @return: tuple of (current, max)
914

915
    """
916
    return self._UnlockedGetMasterCandidateStats()
917

    
918
  @locking.ssynchronized(_config_lock)
919
  def MaintainCandidatePool(self):
920
    """Try to grow the candidate pool to the desired size.
921

922
    @rtype: list
923
    @return: list with the adjusted nodes (L{objects.Node} instances)
924

925
    """
926
    mc_now, mc_max = self._UnlockedGetMasterCandidateStats()
927
    mod_list = []
928
    if mc_now < mc_max:
929
      node_list = self._config_data.nodes.keys()
930
      random.shuffle(node_list)
931
      for name in node_list:
932
        if mc_now >= mc_max:
933
          break
934
        node = self._config_data.nodes[name]
935
        if node.master_candidate or node.offline:
936
          continue
937
        mod_list.append(node)
938
        node.master_candidate = True
939
        node.serial_no += 1
940
        mc_now += 1
941
      if mc_now != mc_max:
942
        # this should not happen
943
        logging.warning("Warning: MaintainCandidatePool didn't manage to"
944
                        " fill the candidate pool (%d/%d)", mc_now, mc_max)
945
      if mod_list:
946
        self._config_data.cluster.serial_no += 1
947
        self._WriteConfig()
948

    
949
    return mod_list
950

    
951
  def _BumpSerialNo(self):
952
    """Bump up the serial number of the config.
953

954
    """
955
    self._config_data.serial_no += 1
956

    
957
  def _OpenConfig(self):
958
    """Read the config data from disk.
959

960
    """
961
    f = open(self._cfg_file, 'r')
962
    try:
963
      try:
964
        data = objects.ConfigData.FromDict(serializer.Load(f.read()))
965
      except Exception, err:
966
        raise errors.ConfigurationError(err)
967
    finally:
968
      f.close()
969

    
970
    # Make sure the configuration has the right version
971
    _ValidateConfig(data)
972

    
973
    if (not hasattr(data, 'cluster') or
974
        not hasattr(data.cluster, 'rsahostkeypub')):
975
      raise errors.ConfigurationError("Incomplete configuration"
976
                                      " (missing cluster.rsahostkeypub)")
977
    self._config_data = data
978
    # reset the last serial as -1 so that the next write will cause
979
    # ssconf update
980
    self._last_cluster_serial = -1
981

    
982
  def _DistributeConfig(self):
983
    """Distribute the configuration to the other nodes.
984

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

988
    """
989
    if self._offline:
990
      return True
991
    bad = False
992

    
993
    node_list = []
994
    addr_list = []
995
    myhostname = self._my_hostname
996
    # we can skip checking whether _UnlockedGetNodeInfo returns None
997
    # since the node list comes from _UnlocketGetNodeList, and we are
998
    # called with the lock held, so no modifications should take place
999
    # in between
1000
    for node_name in self._UnlockedGetNodeList():
1001
      if node_name == myhostname:
1002
        continue
1003
      node_info = self._UnlockedGetNodeInfo(node_name)
1004
      if not node_info.master_candidate:
1005
        continue
1006
      node_list.append(node_info.name)
1007
      addr_list.append(node_info.primary_ip)
1008

    
1009
    result = rpc.RpcRunner.call_upload_file(node_list, self._cfg_file,
1010
                                            address_list=addr_list)
1011
    for node in node_list:
1012
      if not result[node]:
1013
        logging.error("copy of file %s to node %s failed",
1014
                      self._cfg_file, node)
1015
        bad = True
1016
    return not bad
1017

    
1018
  def _WriteConfig(self, destination=None):
1019
    """Write the configuration data to persistent storage.
1020

1021
    """
1022
    config_errors = self._UnlockedVerifyConfig()
1023
    if config_errors:
1024
      raise errors.ConfigurationError("Configuration data is not"
1025
                                      " consistent: %s" %
1026
                                      (", ".join(config_errors)))
1027
    if destination is None:
1028
      destination = self._cfg_file
1029
    self._BumpSerialNo()
1030
    txt = serializer.Dump(self._config_data.ToDict())
1031
    dir_name, file_name = os.path.split(destination)
1032
    fd, name = tempfile.mkstemp('.newconfig', file_name, dir_name)
1033
    f = os.fdopen(fd, 'w')
1034
    try:
1035
      f.write(txt)
1036
      os.fsync(f.fileno())
1037
    finally:
1038
      f.close()
1039
    # we don't need to do os.close(fd) as f.close() did it
1040
    os.rename(name, destination)
1041
    self.write_count += 1
1042

    
1043
    # and redistribute the config file to master candidates
1044
    self._DistributeConfig()
1045

    
1046
    # Write ssconf files on all nodes (including locally)
1047
    if self._last_cluster_serial < self._config_data.cluster.serial_no:
1048
      if not self._offline:
1049
        rpc.RpcRunner.call_write_ssconf_files(self._UnlockedGetNodeList(),
1050
                                              self._UnlockedGetSsconfValues())
1051
      self._last_cluster_serial = self._config_data.cluster.serial_no
1052

    
1053
  def _UnlockedGetSsconfValues(self):
1054
    """Return the values needed by ssconf.
1055

1056
    @rtype: dict
1057
    @return: a dictionary with keys the ssconf names and values their
1058
        associated value
1059

1060
    """
1061
    fn = "\n".join
1062
    node_names = utils.NiceSort(self._UnlockedGetNodeList())
1063
    node_info = [self._UnlockedGetNodeInfo(name) for name in node_names]
1064

    
1065
    off_data = fn(node.name for node in node_info if node.offline)
1066
    mc_data = fn(node.name for node in node_info if node.master_candidate)
1067
    node_data = fn(node_names)
1068

    
1069
    cluster = self._config_data.cluster
1070
    return {
1071
      constants.SS_CLUSTER_NAME: cluster.cluster_name,
1072
      constants.SS_FILE_STORAGE_DIR: cluster.file_storage_dir,
1073
      constants.SS_MASTER_CANDIDATES: mc_data,
1074
      constants.SS_MASTER_IP: cluster.master_ip,
1075
      constants.SS_MASTER_NETDEV: cluster.master_netdev,
1076
      constants.SS_MASTER_NODE: cluster.master_node,
1077
      constants.SS_NODE_LIST: node_data,
1078
      constants.SS_OFFLINE_NODES: off_data,
1079
      constants.SS_RELEASE_VERSION: constants.RELEASE_VERSION,
1080
      }
1081

    
1082
  @locking.ssynchronized(_config_lock)
1083
  def InitConfig(self, version, cluster_config, master_node_config):
1084
    """Create the initial cluster configuration.
1085

1086
    It will contain the current node, which will also be the master
1087
    node, and no instances.
1088

1089
    @type version: int
1090
    @param version: Configuration version
1091
    @type cluster_config: objects.Cluster
1092
    @param cluster_config: Cluster configuration
1093
    @type master_node_config: objects.Node
1094
    @param master_node_config: Master node configuration
1095

1096
    """
1097
    nodes = {
1098
      master_node_config.name: master_node_config,
1099
      }
1100

    
1101
    self._config_data = objects.ConfigData(version=version,
1102
                                           cluster=cluster_config,
1103
                                           nodes=nodes,
1104
                                           instances={},
1105
                                           serial_no=1)
1106
    self._WriteConfig()
1107

    
1108
  @locking.ssynchronized(_config_lock, shared=1)
1109
  def GetVGName(self):
1110
    """Return the volume group name.
1111

1112
    """
1113
    return self._config_data.cluster.volume_group_name
1114

    
1115
  @locking.ssynchronized(_config_lock)
1116
  def SetVGName(self, vg_name):
1117
    """Set the volume group name.
1118

1119
    """
1120
    self._config_data.cluster.volume_group_name = vg_name
1121
    self._config_data.cluster.serial_no += 1
1122
    self._WriteConfig()
1123

    
1124
  @locking.ssynchronized(_config_lock, shared=1)
1125
  def GetDefBridge(self):
1126
    """Return the default bridge.
1127

1128
    """
1129
    return self._config_data.cluster.default_bridge
1130

    
1131
  @locking.ssynchronized(_config_lock, shared=1)
1132
  def GetMACPrefix(self):
1133
    """Return the mac prefix.
1134

1135
    """
1136
    return self._config_data.cluster.mac_prefix
1137

    
1138
  @locking.ssynchronized(_config_lock, shared=1)
1139
  def GetClusterInfo(self):
1140
    """Returns informations about the cluster
1141

1142
    @rtype: L{objects.Cluster}
1143
    @return: the cluster object
1144

1145
    """
1146
    return self._config_data.cluster
1147

    
1148
  @locking.ssynchronized(_config_lock)
1149
  def Update(self, target):
1150
    """Notify function to be called after updates.
1151

1152
    This function must be called when an object (as returned by
1153
    GetInstanceInfo, GetNodeInfo, GetCluster) has been updated and the
1154
    caller wants the modifications saved to the backing store. Note
1155
    that all modified objects will be saved, but the target argument
1156
    is the one the caller wants to ensure that it's saved.
1157

1158
    @param target: an instance of either L{objects.Cluster},
1159
        L{objects.Node} or L{objects.Instance} which is existing in
1160
        the cluster
1161

1162
    """
1163
    if self._config_data is None:
1164
      raise errors.ProgrammerError("Configuration file not read,"
1165
                                   " cannot save.")
1166
    update_serial = False
1167
    if isinstance(target, objects.Cluster):
1168
      test = target == self._config_data.cluster
1169
    elif isinstance(target, objects.Node):
1170
      test = target in self._config_data.nodes.values()
1171
      update_serial = True
1172
    elif isinstance(target, objects.Instance):
1173
      test = target in self._config_data.instances.values()
1174
    else:
1175
      raise errors.ProgrammerError("Invalid object type (%s) passed to"
1176
                                   " ConfigWriter.Update" % type(target))
1177
    if not test:
1178
      raise errors.ConfigurationError("Configuration updated since object"
1179
                                      " has been read or unknown object")
1180
    target.serial_no += 1
1181

    
1182
    if update_serial:
1183
      # for node updates, we need to increase the cluster serial too
1184
      self._config_data.cluster.serial_no += 1
1185

    
1186
    if isinstance(target, objects.Instance):
1187
      self._UnlockedReleaseDRBDMinors(target.name)
1188

    
1189
    self._WriteConfig()