Statistics
| Branch: | Tag: | Revision:

root / lib / config.py @ 23f06b2b

History | View | Annotate | Download (40.5 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
    self._temporary_macs = set()
83
    # Note: in order to prevent errors when resolving our name in
84
    # _DistributeConfig, we compute it here once and reuse it; it's
85
    # better to raise an error before starting to modify the config
86
    # file than after it was modified
87
    self._my_hostname = utils.HostInfo().name
88
    self._last_cluster_serial = -1
89
    self._OpenConfig()
90

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

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

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

103
    This should check the current instances for duplicates.
104

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

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

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

129
    """
130
    all_macs = self._AllMACs()
131
    return mac in all_macs or mac in self._temporary_macs
132

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

137
    This checks the current disks for duplicates.
138

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

    
151
  def _ComputeAllLVs(self):
152
    """Compute the list of all LVs.
153

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

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

166
    This checks the current node, instances and disk names for
167
    duplicates.
168

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

174
    @rtype: string
175
    @return: the unique id
176

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

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

199
    @rtype: list
200
    @return: the list of all MACs
201

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

    
208
    return result
209

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

213
    @rtype: list
214
    @return: the list of all DRBD secrets
215

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

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

    
230
    return result
231

    
232
  def _CheckDiskIDs(self, disk, l_ids, p_ids):
233
    """Compute duplicate disk IDs
234

235
    @type disk: L{objects.Disk}
236
    @param disk: the disk at which to start searching
237
    @type l_ids: list
238
    @param l_ids: list of current logical ids
239
    @type p_ids: list
240
    @param p_ids: list of current physical ids
241
    @rtype: list
242
    @return: a list of error messages
243

244
    """
245
    result = []
246
    if disk.logical_id is not None:
247
      if disk.logical_id in l_ids:
248
        result.append("duplicate logical id %s" % str(disk.logical_id))
249
      else:
250
        l_ids.append(disk.logical_id)
251
    if disk.physical_id is not None:
252
      if disk.physical_id in p_ids:
253
        result.append("duplicate physical id %s" % str(disk.physical_id))
254
      else:
255
        p_ids.append(disk.physical_id)
256

    
257
    if disk.children:
258
      for child in disk.children:
259
        result.extend(self._CheckDiskIDs(child, l_ids, p_ids))
260
    return result
261

    
262
  def _UnlockedVerifyConfig(self):
263
    """Verify function.
264

265
    @rtype: list
266
    @return: a list of error messages; a non-empty list signifies
267
        configuration errors
268

269
    """
270
    result = []
271
    seen_macs = []
272
    ports = {}
273
    data = self._config_data
274
    seen_lids = []
275
    seen_pids = []
276
    for instance_name in data.instances:
277
      instance = data.instances[instance_name]
278
      if instance.primary_node not in data.nodes:
279
        result.append("instance '%s' has invalid primary node '%s'" %
280
                      (instance_name, instance.primary_node))
281
      for snode in instance.secondary_nodes:
282
        if snode not in data.nodes:
283
          result.append("instance '%s' has invalid secondary node '%s'" %
284
                        (instance_name, snode))
285
      for idx, nic in enumerate(instance.nics):
286
        if nic.mac in seen_macs:
287
          result.append("instance '%s' has NIC %d mac %s duplicate" %
288
                        (instance_name, idx, nic.mac))
289
        else:
290
          seen_macs.append(nic.mac)
291

    
292
      # gather the drbd ports for duplicate checks
293
      for dsk in instance.disks:
294
        if dsk.dev_type in constants.LDS_DRBD:
295
          tcp_port = dsk.logical_id[2]
296
          if tcp_port not in ports:
297
            ports[tcp_port] = []
298
          ports[tcp_port].append((instance.name, "drbd disk %s" % dsk.iv_name))
299
      # gather network port reservation
300
      net_port = getattr(instance, "network_port", None)
301
      if net_port is not None:
302
        if net_port not in ports:
303
          ports[net_port] = []
304
        ports[net_port].append((instance.name, "network port"))
305

    
306
      # instance disk verify
307
      for idx, disk in enumerate(instance.disks):
308
        result.extend(["instance '%s' disk %d error: %s" %
309
                       (instance.name, idx, msg) for msg in disk.Verify()])
310
        result.extend(self._CheckDiskIDs(disk, seen_lids, seen_pids))
311

    
312
    # cluster-wide pool of free ports
313
    for free_port in data.cluster.tcpudp_port_pool:
314
      if free_port not in ports:
315
        ports[free_port] = []
316
      ports[free_port].append(("cluster", "port marked as free"))
317

    
318
    # compute tcp/udp duplicate ports
319
    keys = ports.keys()
320
    keys.sort()
321
    for pnum in keys:
322
      pdata = ports[pnum]
323
      if len(pdata) > 1:
324
        txt = ", ".join(["%s/%s" % val for val in pdata])
325
        result.append("tcp/udp port %s has duplicates: %s" % (pnum, txt))
326

    
327
    # highest used tcp port check
328
    if keys:
329
      if keys[-1] > data.cluster.highest_used_port:
330
        result.append("Highest used port mismatch, saved %s, computed %s" %
331
                      (data.cluster.highest_used_port, keys[-1]))
332

    
333
    if not data.nodes[data.cluster.master_node].master_candidate:
334
      result.append("Master node is not a master candidate")
335

    
336
    # master candidate checks
337
    mc_now, mc_max = self._UnlockedGetMasterCandidateStats()
338
    if mc_now < mc_max:
339
      result.append("Not enough master candidates: actual %d, target %d" %
340
                    (mc_now, mc_max))
341

    
342
    # node checks
343
    for node in data.nodes.values():
344
      if [node.master_candidate, node.drained, node.offline].count(True) > 1:
345
        result.append("Node %s state is invalid: master_candidate=%s,"
346
                      " drain=%s, offline=%s" %
347
                      (node.name, node.master_candidate, node.drain,
348
                       node.offline))
349

    
350
    # drbd minors check
351
    d_map, duplicates = self._UnlockedComputeDRBDMap()
352
    for node, minor, instance_a, instance_b in duplicates:
353
      result.append("DRBD minor %d on node %s is assigned twice to instances"
354
                    " %s and %s" % (minor, node, instance_a, instance_b))
355

    
356
    return result
357

    
358
  @locking.ssynchronized(_config_lock, shared=1)
359
  def VerifyConfig(self):
360
    """Verify function.
361

362
    This is just a wrapper over L{_UnlockedVerifyConfig}.
363

364
    @rtype: list
365
    @return: a list of error messages; a non-empty list signifies
366
        configuration errors
367

368
    """
369
    return self._UnlockedVerifyConfig()
370

    
371
  def _UnlockedSetDiskID(self, disk, node_name):
372
    """Convert the unique ID to the ID needed on the target nodes.
373

374
    This is used only for drbd, which needs ip/port configuration.
375

376
    The routine descends down and updates its children also, because
377
    this helps when the only the top device is passed to the remote
378
    node.
379

380
    This function is for internal use, when the config lock is already held.
381

382
    """
383
    if disk.children:
384
      for child in disk.children:
385
        self._UnlockedSetDiskID(child, node_name)
386

    
387
    if disk.logical_id is None and disk.physical_id is not None:
388
      return
389
    if disk.dev_type == constants.LD_DRBD8:
390
      pnode, snode, port, pminor, sminor, secret = disk.logical_id
391
      if node_name not in (pnode, snode):
392
        raise errors.ConfigurationError("DRBD device not knowing node %s" %
393
                                        node_name)
394
      pnode_info = self._UnlockedGetNodeInfo(pnode)
395
      snode_info = self._UnlockedGetNodeInfo(snode)
396
      if pnode_info is None or snode_info is None:
397
        raise errors.ConfigurationError("Can't find primary or secondary node"
398
                                        " for %s" % str(disk))
399
      p_data = (pnode_info.secondary_ip, port)
400
      s_data = (snode_info.secondary_ip, port)
401
      if pnode == node_name:
402
        disk.physical_id = p_data + s_data + (pminor, secret)
403
      else: # it must be secondary, we tested above
404
        disk.physical_id = s_data + p_data + (sminor, secret)
405
    else:
406
      disk.physical_id = disk.logical_id
407
    return
408

    
409
  @locking.ssynchronized(_config_lock)
410
  def SetDiskID(self, disk, node_name):
411
    """Convert the unique ID to the ID needed on the target nodes.
412

413
    This is used only for drbd, which needs ip/port configuration.
414

415
    The routine descends down and updates its children also, because
416
    this helps when the only the top device is passed to the remote
417
    node.
418

419
    """
420
    return self._UnlockedSetDiskID(disk, node_name)
421

    
422
  @locking.ssynchronized(_config_lock)
423
  def AddTcpUdpPort(self, port):
424
    """Adds a new port to the available port pool.
425

426
    """
427
    if not isinstance(port, int):
428
      raise errors.ProgrammerError("Invalid type passed for port")
429

    
430
    self._config_data.cluster.tcpudp_port_pool.add(port)
431
    self._WriteConfig()
432

    
433
  @locking.ssynchronized(_config_lock, shared=1)
434
  def GetPortList(self):
435
    """Returns a copy of the current port list.
436

437
    """
438
    return self._config_data.cluster.tcpudp_port_pool.copy()
439

    
440
  @locking.ssynchronized(_config_lock)
441
  def AllocatePort(self):
442
    """Allocate a port.
443

444
    The port will be taken from the available port pool or from the
445
    default port range (and in this case we increase
446
    highest_used_port).
447

448
    """
449
    # If there are TCP/IP ports configured, we use them first.
450
    if self._config_data.cluster.tcpudp_port_pool:
451
      port = self._config_data.cluster.tcpudp_port_pool.pop()
452
    else:
453
      port = self._config_data.cluster.highest_used_port + 1
454
      if port >= constants.LAST_DRBD_PORT:
455
        raise errors.ConfigurationError("The highest used port is greater"
456
                                        " than %s. Aborting." %
457
                                        constants.LAST_DRBD_PORT)
458
      self._config_data.cluster.highest_used_port = port
459

    
460
    self._WriteConfig()
461
    return port
462

    
463
  def _UnlockedComputeDRBDMap(self):
464
    """Compute the used DRBD minor/nodes.
465

466
    @rtype: (dict, list)
467
    @return: dictionary of node_name: dict of minor: instance_name;
468
        the returned dict will have all the nodes in it (even if with
469
        an empty list), and a list of duplicates; if the duplicates
470
        list is not empty, the configuration is corrupted and its caller
471
        should raise an exception
472

473
    """
474
    def _AppendUsedPorts(instance_name, disk, used):
475
      duplicates = []
476
      if disk.dev_type == constants.LD_DRBD8 and len(disk.logical_id) >= 5:
477
        nodeA, nodeB, dummy, minorA, minorB = disk.logical_id[:5]
478
        for node, port in ((nodeA, minorA), (nodeB, minorB)):
479
          assert node in used, ("Node '%s' of instance '%s' not found"
480
                                " in node list" % (node, instance_name))
481
          if port in used[node]:
482
            duplicates.append((node, port, instance_name, used[node][port]))
483
          else:
484
            used[node][port] = instance_name
485
      if disk.children:
486
        for child in disk.children:
487
          duplicates.extend(_AppendUsedPorts(instance_name, child, used))
488
      return duplicates
489

    
490
    duplicates = []
491
    my_dict = dict((node, {}) for node in self._config_data.nodes)
492
    for instance in self._config_data.instances.itervalues():
493
      for disk in instance.disks:
494
        duplicates.extend(_AppendUsedPorts(instance.name, disk, my_dict))
495
    for (node, minor), instance in self._temporary_drbds.iteritems():
496
      if minor in my_dict[node] and my_dict[node][minor] != instance:
497
        duplicates.append((node, minor, instance, my_dict[node][minor]))
498
      else:
499
        my_dict[node][minor] = instance
500
    return my_dict, duplicates
501

    
502
  @locking.ssynchronized(_config_lock)
503
  def ComputeDRBDMap(self):
504
    """Compute the used DRBD minor/nodes.
505

506
    This is just a wrapper over L{_UnlockedComputeDRBDMap}.
507

508
    @return: dictionary of node_name: dict of minor: instance_name;
509
        the returned dict will have all the nodes in it (even if with
510
        an empty list).
511

512
    """
513
    d_map, duplicates = self._UnlockedComputeDRBDMap()
514
    if duplicates:
515
      raise errors.ConfigurationError("Duplicate DRBD ports detected: %s" %
516
                                      str(duplicates))
517
    return d_map
518

    
519
  @locking.ssynchronized(_config_lock)
520
  def AllocateDRBDMinor(self, nodes, instance):
521
    """Allocate a drbd minor.
522

523
    The free minor will be automatically computed from the existing
524
    devices. A node can be given multiple times in order to allocate
525
    multiple minors. The result is the list of minors, in the same
526
    order as the passed nodes.
527

528
    @type instance: string
529
    @param instance: the instance for which we allocate minors
530

531
    """
532
    assert isinstance(instance, basestring), \
533
           "Invalid argument '%s' passed to AllocateDRBDMinor" % instance
534

    
535
    d_map, duplicates = self._UnlockedComputeDRBDMap()
536
    if duplicates:
537
      raise errors.ConfigurationError("Duplicate DRBD ports detected: %s" %
538
                                      str(duplicates))
539
    result = []
540
    for nname in nodes:
541
      ndata = d_map[nname]
542
      if not ndata:
543
        # no minors used, we can start at 0
544
        result.append(0)
545
        ndata[0] = instance
546
        self._temporary_drbds[(nname, 0)] = instance
547
        continue
548
      keys = ndata.keys()
549
      keys.sort()
550
      ffree = utils.FirstFree(keys)
551
      if ffree is None:
552
        # return the next minor
553
        # TODO: implement high-limit check
554
        minor = keys[-1] + 1
555
      else:
556
        minor = ffree
557
      # double-check minor against current instances
558
      assert minor not in d_map[nname], \
559
             ("Attempt to reuse allocated DRBD minor %d on node %s,"
560
              " already allocated to instance %s" %
561
              (minor, nname, d_map[nname][minor]))
562
      ndata[minor] = instance
563
      # double-check minor against reservation
564
      r_key = (nname, minor)
565
      assert r_key not in self._temporary_drbds, \
566
             ("Attempt to reuse reserved DRBD minor %d on node %s,"
567
              " reserved for instance %s" %
568
              (minor, nname, self._temporary_drbds[r_key]))
569
      self._temporary_drbds[r_key] = instance
570
      result.append(minor)
571
    logging.debug("Request to allocate drbd minors, input: %s, returning %s",
572
                  nodes, result)
573
    return result
574

    
575
  def _UnlockedReleaseDRBDMinors(self, instance):
576
    """Release temporary drbd minors allocated for a given instance.
577

578
    @type instance: string
579
    @param instance: the instance for which temporary minors should be
580
                     released
581

582
    """
583
    assert isinstance(instance, basestring), \
584
           "Invalid argument passed to ReleaseDRBDMinors"
585
    for key, name in self._temporary_drbds.items():
586
      if name == instance:
587
        del self._temporary_drbds[key]
588

    
589
  @locking.ssynchronized(_config_lock)
590
  def ReleaseDRBDMinors(self, instance):
591
    """Release temporary drbd minors allocated for a given instance.
592

593
    This should be called on the error paths, on the success paths
594
    it's automatically called by the ConfigWriter add and update
595
    functions.
596

597
    This function is just a wrapper over L{_UnlockedReleaseDRBDMinors}.
598

599
    @type instance: string
600
    @param instance: the instance for which temporary minors should be
601
                     released
602

603
    """
604
    self._UnlockedReleaseDRBDMinors(instance)
605

    
606
  @locking.ssynchronized(_config_lock, shared=1)
607
  def GetConfigVersion(self):
608
    """Get the configuration version.
609

610
    @return: Config version
611

612
    """
613
    return self._config_data.version
614

    
615
  @locking.ssynchronized(_config_lock, shared=1)
616
  def GetClusterName(self):
617
    """Get cluster name.
618

619
    @return: Cluster name
620

621
    """
622
    return self._config_data.cluster.cluster_name
623

    
624
  @locking.ssynchronized(_config_lock, shared=1)
625
  def GetMasterNode(self):
626
    """Get the hostname of the master node for this cluster.
627

628
    @return: Master hostname
629

630
    """
631
    return self._config_data.cluster.master_node
632

    
633
  @locking.ssynchronized(_config_lock, shared=1)
634
  def GetMasterIP(self):
635
    """Get the IP of the master node for this cluster.
636

637
    @return: Master IP
638

639
    """
640
    return self._config_data.cluster.master_ip
641

    
642
  @locking.ssynchronized(_config_lock, shared=1)
643
  def GetMasterNetdev(self):
644
    """Get the master network device for this cluster.
645

646
    """
647
    return self._config_data.cluster.master_netdev
648

    
649
  @locking.ssynchronized(_config_lock, shared=1)
650
  def GetFileStorageDir(self):
651
    """Get the file storage dir for this cluster.
652

653
    """
654
    return self._config_data.cluster.file_storage_dir
655

    
656
  @locking.ssynchronized(_config_lock, shared=1)
657
  def GetHypervisorType(self):
658
    """Get the hypervisor type for this cluster.
659

660
    """
661
    return self._config_data.cluster.default_hypervisor
662

    
663
  @locking.ssynchronized(_config_lock, shared=1)
664
  def GetHostKey(self):
665
    """Return the rsa hostkey from the config.
666

667
    @rtype: string
668
    @return: the rsa hostkey
669

670
    """
671
    return self._config_data.cluster.rsahostkeypub
672

    
673
  @locking.ssynchronized(_config_lock)
674
  def AddInstance(self, instance):
675
    """Add an instance to the config.
676

677
    This should be used after creating a new instance.
678

679
    @type instance: L{objects.Instance}
680
    @param instance: the instance object
681

682
    """
683
    if not isinstance(instance, objects.Instance):
684
      raise errors.ProgrammerError("Invalid type passed to AddInstance")
685

    
686
    if instance.disk_template != constants.DT_DISKLESS:
687
      all_lvs = instance.MapLVsByNode()
688
      logging.info("Instance '%s' DISK_LAYOUT: %s", instance.name, all_lvs)
689

    
690
    all_macs = self._AllMACs()
691
    for nic in instance.nics:
692
      if nic.mac in all_macs:
693
        raise errors.ConfigurationError("Cannot add instance %s:"
694
          " MAC address '%s' already in use." % (instance.name, nic.mac))
695

    
696
    instance.serial_no = 1
697
    self._config_data.instances[instance.name] = instance
698
    self._config_data.cluster.serial_no += 1
699
    self._UnlockedReleaseDRBDMinors(instance.name)
700
    for nic in instance.nics:
701
      self._temporary_macs.discard(nic.mac)
702
    self._WriteConfig()
703

    
704
  def _SetInstanceStatus(self, instance_name, status):
705
    """Set the instance's status to a given value.
706

707
    """
708
    assert isinstance(status, bool), \
709
           "Invalid status '%s' passed to SetInstanceStatus" % (status,)
710

    
711
    if instance_name not in self._config_data.instances:
712
      raise errors.ConfigurationError("Unknown instance '%s'" %
713
                                      instance_name)
714
    instance = self._config_data.instances[instance_name]
715
    if instance.admin_up != status:
716
      instance.admin_up = status
717
      instance.serial_no += 1
718
      self._WriteConfig()
719

    
720
  @locking.ssynchronized(_config_lock)
721
  def MarkInstanceUp(self, instance_name):
722
    """Mark the instance status to up in the config.
723

724
    """
725
    self._SetInstanceStatus(instance_name, True)
726

    
727
  @locking.ssynchronized(_config_lock)
728
  def RemoveInstance(self, instance_name):
729
    """Remove the instance from the configuration.
730

731
    """
732
    if instance_name not in self._config_data.instances:
733
      raise errors.ConfigurationError("Unknown instance '%s'" % instance_name)
734
    del self._config_data.instances[instance_name]
735
    self._config_data.cluster.serial_no += 1
736
    self._WriteConfig()
737

    
738
  @locking.ssynchronized(_config_lock)
739
  def RenameInstance(self, old_name, new_name):
740
    """Rename an instance.
741

742
    This needs to be done in ConfigWriter and not by RemoveInstance
743
    combined with AddInstance as only we can guarantee an atomic
744
    rename.
745

746
    """
747
    if old_name not in self._config_data.instances:
748
      raise errors.ConfigurationError("Unknown instance '%s'" % old_name)
749
    inst = self._config_data.instances[old_name]
750
    del self._config_data.instances[old_name]
751
    inst.name = new_name
752

    
753
    for disk in inst.disks:
754
      if disk.dev_type == constants.LD_FILE:
755
        # rename the file paths in logical and physical id
756
        file_storage_dir = os.path.dirname(os.path.dirname(disk.logical_id[1]))
757
        disk.physical_id = disk.logical_id = (disk.logical_id[0],
758
                                              os.path.join(file_storage_dir,
759
                                                           inst.name,
760
                                                           disk.iv_name))
761

    
762
    self._config_data.instances[inst.name] = inst
763
    self._WriteConfig()
764

    
765
  @locking.ssynchronized(_config_lock)
766
  def MarkInstanceDown(self, instance_name):
767
    """Mark the status of an instance to down in the configuration.
768

769
    """
770
    self._SetInstanceStatus(instance_name, False)
771

    
772
  def _UnlockedGetInstanceList(self):
773
    """Get the list of instances.
774

775
    This function is for internal use, when the config lock is already held.
776

777
    """
778
    return self._config_data.instances.keys()
779

    
780
  @locking.ssynchronized(_config_lock, shared=1)
781
  def GetInstanceList(self):
782
    """Get the list of instances.
783

784
    @return: array of instances, ex. ['instance2.example.com',
785
        'instance1.example.com']
786

787
    """
788
    return self._UnlockedGetInstanceList()
789

    
790
  @locking.ssynchronized(_config_lock, shared=1)
791
  def ExpandInstanceName(self, short_name):
792
    """Attempt to expand an incomplete instance name.
793

794
    """
795
    return utils.MatchNameComponent(short_name,
796
                                    self._config_data.instances.keys())
797

    
798
  def _UnlockedGetInstanceInfo(self, instance_name):
799
    """Returns informations about an instance.
800

801
    This function is for internal use, when the config lock is already held.
802

803
    """
804
    if instance_name not in self._config_data.instances:
805
      return None
806

    
807
    return self._config_data.instances[instance_name]
808

    
809
  @locking.ssynchronized(_config_lock, shared=1)
810
  def GetInstanceInfo(self, instance_name):
811
    """Returns informations about an instance.
812

813
    It takes the information from the configuration file. Other informations of
814
    an instance are taken from the live systems.
815

816
    @param instance_name: name of the instance, e.g.
817
        I{instance1.example.com}
818

819
    @rtype: L{objects.Instance}
820
    @return: the instance object
821

822
    """
823
    return self._UnlockedGetInstanceInfo(instance_name)
824

    
825
  @locking.ssynchronized(_config_lock, shared=1)
826
  def GetAllInstancesInfo(self):
827
    """Get the configuration of all instances.
828

829
    @rtype: dict
830
    @return: dict of (instance, instance_info), where instance_info is what
831
              would GetInstanceInfo return for the node
832

833
    """
834
    my_dict = dict([(instance, self._UnlockedGetInstanceInfo(instance))
835
                    for instance in self._UnlockedGetInstanceList()])
836
    return my_dict
837

    
838
  @locking.ssynchronized(_config_lock)
839
  def AddNode(self, node):
840
    """Add a node to the configuration.
841

842
    @type node: L{objects.Node}
843
    @param node: a Node instance
844

845
    """
846
    logging.info("Adding node %s to configuration" % node.name)
847

    
848
    node.serial_no = 1
849
    self._config_data.nodes[node.name] = node
850
    self._config_data.cluster.serial_no += 1
851
    self._WriteConfig()
852

    
853
  @locking.ssynchronized(_config_lock)
854
  def RemoveNode(self, node_name):
855
    """Remove a node from the configuration.
856

857
    """
858
    logging.info("Removing node %s from configuration" % node_name)
859

    
860
    if node_name not in self._config_data.nodes:
861
      raise errors.ConfigurationError("Unknown node '%s'" % node_name)
862

    
863
    del self._config_data.nodes[node_name]
864
    self._config_data.cluster.serial_no += 1
865
    self._WriteConfig()
866

    
867
  @locking.ssynchronized(_config_lock, shared=1)
868
  def ExpandNodeName(self, short_name):
869
    """Attempt to expand an incomplete instance name.
870

871
    """
872
    return utils.MatchNameComponent(short_name,
873
                                    self._config_data.nodes.keys())
874

    
875
  def _UnlockedGetNodeInfo(self, node_name):
876
    """Get the configuration of a node, as stored in the config.
877

878
    This function is for internal use, when the config lock is already
879
    held.
880

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

883
    @rtype: L{objects.Node}
884
    @return: the node object
885

886
    """
887
    if node_name not in self._config_data.nodes:
888
      return None
889

    
890
    return self._config_data.nodes[node_name]
891

    
892

    
893
  @locking.ssynchronized(_config_lock, shared=1)
894
  def GetNodeInfo(self, node_name):
895
    """Get the configuration of a node, as stored in the config.
896

897
    This is just a locked wrapper over L{_UnlockedGetNodeInfo}.
898

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

901
    @rtype: L{objects.Node}
902
    @return: the node object
903

904
    """
905
    return self._UnlockedGetNodeInfo(node_name)
906

    
907
  def _UnlockedGetNodeList(self):
908
    """Return the list of nodes which are in the configuration.
909

910
    This function is for internal use, when the config lock is already
911
    held.
912

913
    @rtype: list
914

915
    """
916
    return self._config_data.nodes.keys()
917

    
918

    
919
  @locking.ssynchronized(_config_lock, shared=1)
920
  def GetNodeList(self):
921
    """Return the list of nodes which are in the configuration.
922

923
    """
924
    return self._UnlockedGetNodeList()
925

    
926
  @locking.ssynchronized(_config_lock, shared=1)
927
  def GetOnlineNodeList(self):
928
    """Return the list of nodes which are online.
929

930
    """
931
    all_nodes = [self._UnlockedGetNodeInfo(node)
932
                 for node in self._UnlockedGetNodeList()]
933
    return [node.name for node in all_nodes if not node.offline]
934

    
935
  @locking.ssynchronized(_config_lock, shared=1)
936
  def GetAllNodesInfo(self):
937
    """Get the configuration of all nodes.
938

939
    @rtype: dict
940
    @return: dict of (node, node_info), where node_info is what
941
              would GetNodeInfo return for the node
942

943
    """
944
    my_dict = dict([(node, self._UnlockedGetNodeInfo(node))
945
                    for node in self._UnlockedGetNodeList()])
946
    return my_dict
947

    
948
  def _UnlockedGetMasterCandidateStats(self, exceptions=None):
949
    """Get the number of current and maximum desired and possible candidates.
950

951
    @type exceptions: list
952
    @param exceptions: if passed, list of nodes that should be ignored
953
    @rtype: tuple
954
    @return: tuple of (current, desired and possible)
955

956
    """
957
    mc_now = mc_max = 0
958
    for node in self._config_data.nodes.values():
959
      if exceptions and node.name in exceptions:
960
        continue
961
      if not (node.offline or node.drained):
962
        mc_max += 1
963
      if node.master_candidate:
964
        mc_now += 1
965
    mc_max = min(mc_max, self._config_data.cluster.candidate_pool_size)
966
    return (mc_now, mc_max)
967

    
968
  @locking.ssynchronized(_config_lock, shared=1)
969
  def GetMasterCandidateStats(self, exceptions=None):
970
    """Get the number of current and maximum possible candidates.
971

972
    This is just a wrapper over L{_UnlockedGetMasterCandidateStats}.
973

974
    @type exceptions: list
975
    @param exceptions: if passed, list of nodes that should be ignored
976
    @rtype: tuple
977
    @return: tuple of (current, max)
978

979
    """
980
    return self._UnlockedGetMasterCandidateStats(exceptions)
981

    
982
  @locking.ssynchronized(_config_lock)
983
  def MaintainCandidatePool(self):
984
    """Try to grow the candidate pool to the desired size.
985

986
    @rtype: list
987
    @return: list with the adjusted nodes (L{objects.Node} instances)
988

989
    """
990
    mc_now, mc_max = self._UnlockedGetMasterCandidateStats()
991
    mod_list = []
992
    if mc_now < mc_max:
993
      node_list = self._config_data.nodes.keys()
994
      random.shuffle(node_list)
995
      for name in node_list:
996
        if mc_now >= mc_max:
997
          break
998
        node = self._config_data.nodes[name]
999
        if node.master_candidate or node.offline or node.drained:
1000
          continue
1001
        mod_list.append(node)
1002
        node.master_candidate = True
1003
        node.serial_no += 1
1004
        mc_now += 1
1005
      if mc_now != mc_max:
1006
        # this should not happen
1007
        logging.warning("Warning: MaintainCandidatePool didn't manage to"
1008
                        " fill the candidate pool (%d/%d)", mc_now, mc_max)
1009
      if mod_list:
1010
        self._config_data.cluster.serial_no += 1
1011
        self._WriteConfig()
1012

    
1013
    return mod_list
1014

    
1015
  def _BumpSerialNo(self):
1016
    """Bump up the serial number of the config.
1017

1018
    """
1019
    self._config_data.serial_no += 1
1020

    
1021
  def _OpenConfig(self):
1022
    """Read the config data from disk.
1023

1024
    """
1025
    f = open(self._cfg_file, 'r')
1026
    try:
1027
      try:
1028
        data = objects.ConfigData.FromDict(serializer.Load(f.read()))
1029
      except Exception, err:
1030
        raise errors.ConfigurationError(err)
1031
    finally:
1032
      f.close()
1033

    
1034
    # Make sure the configuration has the right version
1035
    _ValidateConfig(data)
1036

    
1037
    if (not hasattr(data, 'cluster') or
1038
        not hasattr(data.cluster, 'rsahostkeypub')):
1039
      raise errors.ConfigurationError("Incomplete configuration"
1040
                                      " (missing cluster.rsahostkeypub)")
1041
    self._config_data = data
1042
    # reset the last serial as -1 so that the next write will cause
1043
    # ssconf update
1044
    self._last_cluster_serial = -1
1045

    
1046
  def _DistributeConfig(self):
1047
    """Distribute the configuration to the other nodes.
1048

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

1052
    """
1053
    if self._offline:
1054
      return True
1055
    bad = False
1056

    
1057
    node_list = []
1058
    addr_list = []
1059
    myhostname = self._my_hostname
1060
    # we can skip checking whether _UnlockedGetNodeInfo returns None
1061
    # since the node list comes from _UnlocketGetNodeList, and we are
1062
    # called with the lock held, so no modifications should take place
1063
    # in between
1064
    for node_name in self._UnlockedGetNodeList():
1065
      if node_name == myhostname:
1066
        continue
1067
      node_info = self._UnlockedGetNodeInfo(node_name)
1068
      if not node_info.master_candidate:
1069
        continue
1070
      node_list.append(node_info.name)
1071
      addr_list.append(node_info.primary_ip)
1072

    
1073
    result = rpc.RpcRunner.call_upload_file(node_list, self._cfg_file,
1074
                                            address_list=addr_list)
1075
    for node in node_list:
1076
      if not result[node]:
1077
        logging.error("copy of file %s to node %s failed",
1078
                      self._cfg_file, node)
1079
        bad = True
1080
    return not bad
1081

    
1082
  def _WriteConfig(self, destination=None):
1083
    """Write the configuration data to persistent storage.
1084

1085
    """
1086
    config_errors = self._UnlockedVerifyConfig()
1087
    if config_errors:
1088
      raise errors.ConfigurationError("Configuration data is not"
1089
                                      " consistent: %s" %
1090
                                      (", ".join(config_errors)))
1091
    if destination is None:
1092
      destination = self._cfg_file
1093
    self._BumpSerialNo()
1094
    txt = serializer.Dump(self._config_data.ToDict())
1095
    dir_name, file_name = os.path.split(destination)
1096
    fd, name = tempfile.mkstemp('.newconfig', file_name, dir_name)
1097
    f = os.fdopen(fd, 'w')
1098
    try:
1099
      f.write(txt)
1100
      os.fsync(f.fileno())
1101
    finally:
1102
      f.close()
1103
    # we don't need to do os.close(fd) as f.close() did it
1104
    os.rename(name, destination)
1105
    self.write_count += 1
1106

    
1107
    # and redistribute the config file to master candidates
1108
    self._DistributeConfig()
1109

    
1110
    # Write ssconf files on all nodes (including locally)
1111
    if self._last_cluster_serial < self._config_data.cluster.serial_no:
1112
      if not self._offline:
1113
        rpc.RpcRunner.call_write_ssconf_files(self._UnlockedGetNodeList(),
1114
                                              self._UnlockedGetSsconfValues())
1115
      self._last_cluster_serial = self._config_data.cluster.serial_no
1116

    
1117
  def _UnlockedGetSsconfValues(self):
1118
    """Return the values needed by ssconf.
1119

1120
    @rtype: dict
1121
    @return: a dictionary with keys the ssconf names and values their
1122
        associated value
1123

1124
    """
1125
    fn = "\n".join
1126
    instance_names = utils.NiceSort(self._UnlockedGetInstanceList())
1127
    node_names = utils.NiceSort(self._UnlockedGetNodeList())
1128
    node_info = [self._UnlockedGetNodeInfo(name) for name in node_names]
1129

    
1130
    instance_data = fn(instance_names)
1131
    off_data = fn(node.name for node in node_info if node.offline)
1132
    on_data = fn(node.name for node in node_info if not node.offline)
1133
    mc_data = fn(node.name for node in node_info if node.master_candidate)
1134
    node_data = fn(node_names)
1135

    
1136
    cluster = self._config_data.cluster
1137
    cluster_tags = fn(cluster.GetTags())
1138
    return {
1139
      constants.SS_CLUSTER_NAME: cluster.cluster_name,
1140
      constants.SS_CLUSTER_TAGS: cluster_tags,
1141
      constants.SS_FILE_STORAGE_DIR: cluster.file_storage_dir,
1142
      constants.SS_MASTER_CANDIDATES: mc_data,
1143
      constants.SS_MASTER_IP: cluster.master_ip,
1144
      constants.SS_MASTER_NETDEV: cluster.master_netdev,
1145
      constants.SS_MASTER_NODE: cluster.master_node,
1146
      constants.SS_NODE_LIST: node_data,
1147
      constants.SS_OFFLINE_NODES: off_data,
1148
      constants.SS_ONLINE_NODES: on_data,
1149
      constants.SS_INSTANCE_LIST: instance_data,
1150
      constants.SS_RELEASE_VERSION: constants.RELEASE_VERSION,
1151
      }
1152

    
1153
  @locking.ssynchronized(_config_lock)
1154
  def InitConfig(self, version, cluster_config, master_node_config):
1155
    """Create the initial cluster configuration.
1156

1157
    It will contain the current node, which will also be the master
1158
    node, and no instances.
1159

1160
    @type version: int
1161
    @param version: Configuration version
1162
    @type cluster_config: objects.Cluster
1163
    @param cluster_config: Cluster configuration
1164
    @type master_node_config: objects.Node
1165
    @param master_node_config: Master node configuration
1166

1167
    """
1168
    nodes = {
1169
      master_node_config.name: master_node_config,
1170
      }
1171

    
1172
    self._config_data = objects.ConfigData(version=version,
1173
                                           cluster=cluster_config,
1174
                                           nodes=nodes,
1175
                                           instances={},
1176
                                           serial_no=1)
1177
    self._WriteConfig()
1178

    
1179
  @locking.ssynchronized(_config_lock, shared=1)
1180
  def GetVGName(self):
1181
    """Return the volume group name.
1182

1183
    """
1184
    return self._config_data.cluster.volume_group_name
1185

    
1186
  @locking.ssynchronized(_config_lock)
1187
  def SetVGName(self, vg_name):
1188
    """Set the volume group name.
1189

1190
    """
1191
    self._config_data.cluster.volume_group_name = vg_name
1192
    self._config_data.cluster.serial_no += 1
1193
    self._WriteConfig()
1194

    
1195
  @locking.ssynchronized(_config_lock, shared=1)
1196
  def GetDefBridge(self):
1197
    """Return the default bridge.
1198

1199
    """
1200
    return self._config_data.cluster.default_bridge
1201

    
1202
  @locking.ssynchronized(_config_lock, shared=1)
1203
  def GetMACPrefix(self):
1204
    """Return the mac prefix.
1205

1206
    """
1207
    return self._config_data.cluster.mac_prefix
1208

    
1209
  @locking.ssynchronized(_config_lock, shared=1)
1210
  def GetClusterInfo(self):
1211
    """Returns informations about the cluster
1212

1213
    @rtype: L{objects.Cluster}
1214
    @return: the cluster object
1215

1216
    """
1217
    return self._config_data.cluster
1218

    
1219
  @locking.ssynchronized(_config_lock)
1220
  def Update(self, target):
1221
    """Notify function to be called after updates.
1222

1223
    This function must be called when an object (as returned by
1224
    GetInstanceInfo, GetNodeInfo, GetCluster) has been updated and the
1225
    caller wants the modifications saved to the backing store. Note
1226
    that all modified objects will be saved, but the target argument
1227
    is the one the caller wants to ensure that it's saved.
1228

1229
    @param target: an instance of either L{objects.Cluster},
1230
        L{objects.Node} or L{objects.Instance} which is existing in
1231
        the cluster
1232

1233
    """
1234
    if self._config_data is None:
1235
      raise errors.ProgrammerError("Configuration file not read,"
1236
                                   " cannot save.")
1237
    update_serial = False
1238
    if isinstance(target, objects.Cluster):
1239
      test = target == self._config_data.cluster
1240
    elif isinstance(target, objects.Node):
1241
      test = target in self._config_data.nodes.values()
1242
      update_serial = True
1243
    elif isinstance(target, objects.Instance):
1244
      test = target in self._config_data.instances.values()
1245
    else:
1246
      raise errors.ProgrammerError("Invalid object type (%s) passed to"
1247
                                   " ConfigWriter.Update" % type(target))
1248
    if not test:
1249
      raise errors.ConfigurationError("Configuration updated since object"
1250
                                      " has been read or unknown object")
1251
    target.serial_no += 1
1252

    
1253
    if update_serial:
1254
      # for node updates, we need to increase the cluster serial too
1255
      self._config_data.cluster.serial_no += 1
1256

    
1257
    if isinstance(target, objects.Instance):
1258
      self._UnlockedReleaseDRBDMinors(target.name)
1259
      for nic in target.nics:
1260
        self._temporary_macs.discard(nic.mac)
1261

    
1262
    self._WriteConfig()