Statistics
| Branch: | Tag: | Revision:

root / lib / config.py @ 44485f49

History | View | Annotate | Download (43.3 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
import time
39

    
40
from ganeti import errors
41
from ganeti import locking
42
from ganeti import utils
43
from ganeti import constants
44
from ganeti import rpc
45
from ganeti import objects
46
from ganeti import serializer
47

    
48

    
49
_config_lock = locking.SharedLock()
50

    
51

    
52
def _ValidateConfig(data):
53
  """Verifies that a configuration objects looks valid.
54

55
  This only verifies the version of the configuration.
56

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

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

    
67

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

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

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

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

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

104
    This should check the current instances for duplicates.
105

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

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

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

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

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

138
    This checks the current disks for duplicates.
139

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

    
152
  def _AllLVs(self):
153
    """Compute the list of all LVs.
154

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

    
163
  def _AllIDs(self, include_temporary):
164
    """Compute the list of all UUIDs and names we have.
165

166
    @type include_temporary: boolean
167
    @param include_temporary: whether to include the _temporary_ids set
168
    @rtype: set
169
    @return: a set of IDs
170

171
    """
172
    existing = set()
173
    if include_temporary:
174
      existing.update(self._temporary_ids)
175
    existing.update(self._AllLVs())
176
    existing.update(self._config_data.instances.keys())
177
    existing.update(self._config_data.nodes.keys())
178
    existing.update([i.uuid for i in self._AllUUIDObjects() if i.uuid])
179
    return existing
180

    
181
  def _GenerateUniqueID(self, exceptions=None):
182
    """Generate an unique UUID.
183

184
    This checks the current node, instances and disk names for
185
    duplicates.
186

187
    @param exceptions: a list with some other names which should be
188
        checked for uniqueness (used for example when you want to get
189
        more than one id at one time without adding each one in turn
190
        to the config file)
191

192
    @rtype: string
193
    @return: the unique id
194

195
    """
196
    existing = self._AllIDs(include_temporary=True)
197
    if exceptions is not None:
198
      existing.update(exceptions)
199
    retries = 64
200
    while retries > 0:
201
      unique_id = utils.NewUUID()
202
      if unique_id not in existing and unique_id is not None:
203
        break
204
    else:
205
      raise errors.ConfigurationError("Not able generate an unique ID"
206
                                      " (last tried ID: %s" % unique_id)
207
    self._temporary_ids.add(unique_id)
208
    return unique_id
209

    
210
  @locking.ssynchronized(_config_lock, shared=1)
211
  def GenerateUniqueID(self, exceptions=None):
212
    """Generate an unique ID.
213

214
    This is just a wrapper over the unlocked version.
215

216
    """
217
    return self._GenerateUniqueID(exceptions=exceptions)
218

    
219
  def _CleanupTemporaryIDs(self):
220
    """Cleanups the _temporary_ids structure.
221

222
    """
223
    existing = self._AllIDs(include_temporary=False)
224
    self._temporary_ids = self._temporary_ids - existing
225

    
226
  def _AllMACs(self):
227
    """Return all MACs present in the config.
228

229
    @rtype: list
230
    @return: the list of all MACs
231

232
    """
233
    result = []
234
    for instance in self._config_data.instances.values():
235
      for nic in instance.nics:
236
        result.append(nic.mac)
237

    
238
    return result
239

    
240
  def _AllDRBDSecrets(self):
241
    """Return all DRBD secrets present in the config.
242

243
    @rtype: list
244
    @return: the list of all DRBD secrets
245

246
    """
247
    def helper(disk, result):
248
      """Recursively gather secrets from this disk."""
249
      if disk.dev_type == constants.DT_DRBD8:
250
        result.append(disk.logical_id[5])
251
      if disk.children:
252
        for child in disk.children:
253
          helper(child, result)
254

    
255
    result = []
256
    for instance in self._config_data.instances.values():
257
      for disk in instance.disks:
258
        helper(disk, result)
259

    
260
    return result
261

    
262
  def _CheckDiskIDs(self, disk, l_ids, p_ids):
263
    """Compute duplicate disk IDs
264

265
    @type disk: L{objects.Disk}
266
    @param disk: the disk at which to start searching
267
    @type l_ids: list
268
    @param l_ids: list of current logical ids
269
    @type p_ids: list
270
    @param p_ids: list of current physical ids
271
    @rtype: list
272
    @return: a list of error messages
273

274
    """
275
    result = []
276
    if disk.logical_id is not None:
277
      if disk.logical_id in l_ids:
278
        result.append("duplicate logical id %s" % str(disk.logical_id))
279
      else:
280
        l_ids.append(disk.logical_id)
281
    if disk.physical_id is not None:
282
      if disk.physical_id in p_ids:
283
        result.append("duplicate physical id %s" % str(disk.physical_id))
284
      else:
285
        p_ids.append(disk.physical_id)
286

    
287
    if disk.children:
288
      for child in disk.children:
289
        result.extend(self._CheckDiskIDs(child, l_ids, p_ids))
290
    return result
291

    
292
  def _UnlockedVerifyConfig(self):
293
    """Verify function.
294

295
    @rtype: list
296
    @return: a list of error messages; a non-empty list signifies
297
        configuration errors
298

299
    """
300
    result = []
301
    seen_macs = []
302
    ports = {}
303
    data = self._config_data
304
    seen_lids = []
305
    seen_pids = []
306

    
307
    # global cluster checks
308
    if not data.cluster.enabled_hypervisors:
309
      result.append("enabled hypervisors list doesn't have any entries")
310
    invalid_hvs = set(data.cluster.enabled_hypervisors) - constants.HYPER_TYPES
311
    if invalid_hvs:
312
      result.append("enabled hypervisors contains invalid entries: %s" %
313
                    invalid_hvs)
314

    
315
    if data.cluster.master_node not in data.nodes:
316
      result.append("cluster has invalid primary node '%s'" %
317
                    data.cluster.master_node)
318

    
319
    # per-instance checks
320
    for instance_name in data.instances:
321
      instance = data.instances[instance_name]
322
      if instance.primary_node not in data.nodes:
323
        result.append("instance '%s' has invalid primary node '%s'" %
324
                      (instance_name, instance.primary_node))
325
      for snode in instance.secondary_nodes:
326
        if snode not in data.nodes:
327
          result.append("instance '%s' has invalid secondary node '%s'" %
328
                        (instance_name, snode))
329
      for idx, nic in enumerate(instance.nics):
330
        if nic.mac in seen_macs:
331
          result.append("instance '%s' has NIC %d mac %s duplicate" %
332
                        (instance_name, idx, nic.mac))
333
        else:
334
          seen_macs.append(nic.mac)
335

    
336
      # gather the drbd ports for duplicate checks
337
      for dsk in instance.disks:
338
        if dsk.dev_type in constants.LDS_DRBD:
339
          tcp_port = dsk.logical_id[2]
340
          if tcp_port not in ports:
341
            ports[tcp_port] = []
342
          ports[tcp_port].append((instance.name, "drbd disk %s" % dsk.iv_name))
343
      # gather network port reservation
344
      net_port = getattr(instance, "network_port", None)
345
      if net_port is not None:
346
        if net_port not in ports:
347
          ports[net_port] = []
348
        ports[net_port].append((instance.name, "network port"))
349

    
350
      # instance disk verify
351
      for idx, disk in enumerate(instance.disks):
352
        result.extend(["instance '%s' disk %d error: %s" %
353
                       (instance.name, idx, msg) for msg in disk.Verify()])
354
        result.extend(self._CheckDiskIDs(disk, seen_lids, seen_pids))
355

    
356
    # cluster-wide pool of free ports
357
    for free_port in data.cluster.tcpudp_port_pool:
358
      if free_port not in ports:
359
        ports[free_port] = []
360
      ports[free_port].append(("cluster", "port marked as free"))
361

    
362
    # compute tcp/udp duplicate ports
363
    keys = ports.keys()
364
    keys.sort()
365
    for pnum in keys:
366
      pdata = ports[pnum]
367
      if len(pdata) > 1:
368
        txt = ", ".join(["%s/%s" % val for val in pdata])
369
        result.append("tcp/udp port %s has duplicates: %s" % (pnum, txt))
370

    
371
    # highest used tcp port check
372
    if keys:
373
      if keys[-1] > data.cluster.highest_used_port:
374
        result.append("Highest used port mismatch, saved %s, computed %s" %
375
                      (data.cluster.highest_used_port, keys[-1]))
376

    
377
    if not data.nodes[data.cluster.master_node].master_candidate:
378
      result.append("Master node is not a master candidate")
379

    
380
    # master candidate checks
381
    mc_now, mc_max, _ = self._UnlockedGetMasterCandidateStats()
382
    if mc_now < mc_max:
383
      result.append("Not enough master candidates: actual %d, target %d" %
384
                    (mc_now, mc_max))
385

    
386
    # node checks
387
    for node in data.nodes.values():
388
      if [node.master_candidate, node.drained, node.offline].count(True) > 1:
389
        result.append("Node %s state is invalid: master_candidate=%s,"
390
                      " drain=%s, offline=%s" %
391
                      (node.name, node.master_candidate, node.drain,
392
                       node.offline))
393

    
394
    # drbd minors check
395
    d_map, duplicates = self._UnlockedComputeDRBDMap()
396
    for node, minor, instance_a, instance_b in duplicates:
397
      result.append("DRBD minor %d on node %s is assigned twice to instances"
398
                    " %s and %s" % (minor, node, instance_a, instance_b))
399

    
400
    return result
401

    
402
  @locking.ssynchronized(_config_lock, shared=1)
403
  def VerifyConfig(self):
404
    """Verify function.
405

406
    This is just a wrapper over L{_UnlockedVerifyConfig}.
407

408
    @rtype: list
409
    @return: a list of error messages; a non-empty list signifies
410
        configuration errors
411

412
    """
413
    return self._UnlockedVerifyConfig()
414

    
415
  def _UnlockedSetDiskID(self, disk, node_name):
416
    """Convert the unique ID to the ID needed on the target nodes.
417

418
    This is used only for drbd, which needs ip/port configuration.
419

420
    The routine descends down and updates its children also, because
421
    this helps when the only the top device is passed to the remote
422
    node.
423

424
    This function is for internal use, when the config lock is already held.
425

426
    """
427
    if disk.children:
428
      for child in disk.children:
429
        self._UnlockedSetDiskID(child, node_name)
430

    
431
    if disk.logical_id is None and disk.physical_id is not None:
432
      return
433
    if disk.dev_type == constants.LD_DRBD8:
434
      pnode, snode, port, pminor, sminor, secret = disk.logical_id
435
      if node_name not in (pnode, snode):
436
        raise errors.ConfigurationError("DRBD device not knowing node %s" %
437
                                        node_name)
438
      pnode_info = self._UnlockedGetNodeInfo(pnode)
439
      snode_info = self._UnlockedGetNodeInfo(snode)
440
      if pnode_info is None or snode_info is None:
441
        raise errors.ConfigurationError("Can't find primary or secondary node"
442
                                        " for %s" % str(disk))
443
      p_data = (pnode_info.secondary_ip, port)
444
      s_data = (snode_info.secondary_ip, port)
445
      if pnode == node_name:
446
        disk.physical_id = p_data + s_data + (pminor, secret)
447
      else: # it must be secondary, we tested above
448
        disk.physical_id = s_data + p_data + (sminor, secret)
449
    else:
450
      disk.physical_id = disk.logical_id
451
    return
452

    
453
  @locking.ssynchronized(_config_lock)
454
  def SetDiskID(self, disk, node_name):
455
    """Convert the unique ID to the ID needed on the target nodes.
456

457
    This is used only for drbd, which needs ip/port configuration.
458

459
    The routine descends down and updates its children also, because
460
    this helps when the only the top device is passed to the remote
461
    node.
462

463
    """
464
    return self._UnlockedSetDiskID(disk, node_name)
465

    
466
  @locking.ssynchronized(_config_lock)
467
  def AddTcpUdpPort(self, port):
468
    """Adds a new port to the available port pool.
469

470
    """
471
    if not isinstance(port, int):
472
      raise errors.ProgrammerError("Invalid type passed for port")
473

    
474
    self._config_data.cluster.tcpudp_port_pool.add(port)
475
    self._WriteConfig()
476

    
477
  @locking.ssynchronized(_config_lock, shared=1)
478
  def GetPortList(self):
479
    """Returns a copy of the current port list.
480

481
    """
482
    return self._config_data.cluster.tcpudp_port_pool.copy()
483

    
484
  @locking.ssynchronized(_config_lock)
485
  def AllocatePort(self):
486
    """Allocate a port.
487

488
    The port will be taken from the available port pool or from the
489
    default port range (and in this case we increase
490
    highest_used_port).
491

492
    """
493
    # If there are TCP/IP ports configured, we use them first.
494
    if self._config_data.cluster.tcpudp_port_pool:
495
      port = self._config_data.cluster.tcpudp_port_pool.pop()
496
    else:
497
      port = self._config_data.cluster.highest_used_port + 1
498
      if port >= constants.LAST_DRBD_PORT:
499
        raise errors.ConfigurationError("The highest used port is greater"
500
                                        " than %s. Aborting." %
501
                                        constants.LAST_DRBD_PORT)
502
      self._config_data.cluster.highest_used_port = port
503

    
504
    self._WriteConfig()
505
    return port
506

    
507
  def _UnlockedComputeDRBDMap(self):
508
    """Compute the used DRBD minor/nodes.
509

510
    @rtype: (dict, list)
511
    @return: dictionary of node_name: dict of minor: instance_name;
512
        the returned dict will have all the nodes in it (even if with
513
        an empty list), and a list of duplicates; if the duplicates
514
        list is not empty, the configuration is corrupted and its caller
515
        should raise an exception
516

517
    """
518
    def _AppendUsedPorts(instance_name, disk, used):
519
      duplicates = []
520
      if disk.dev_type == constants.LD_DRBD8 and len(disk.logical_id) >= 5:
521
        node_a, node_b, _, minor_a, minor_b = disk.logical_id[:5]
522
        for node, port in ((node_a, minor_a), (node_b, minor_b)):
523
          assert node in used, ("Node '%s' of instance '%s' not found"
524
                                " in node list" % (node, instance_name))
525
          if port in used[node]:
526
            duplicates.append((node, port, instance_name, used[node][port]))
527
          else:
528
            used[node][port] = instance_name
529
      if disk.children:
530
        for child in disk.children:
531
          duplicates.extend(_AppendUsedPorts(instance_name, child, used))
532
      return duplicates
533

    
534
    duplicates = []
535
    my_dict = dict((node, {}) for node in self._config_data.nodes)
536
    for instance in self._config_data.instances.itervalues():
537
      for disk in instance.disks:
538
        duplicates.extend(_AppendUsedPorts(instance.name, disk, my_dict))
539
    for (node, minor), instance in self._temporary_drbds.iteritems():
540
      if minor in my_dict[node] and my_dict[node][minor] != instance:
541
        duplicates.append((node, minor, instance, my_dict[node][minor]))
542
      else:
543
        my_dict[node][minor] = instance
544
    return my_dict, duplicates
545

    
546
  @locking.ssynchronized(_config_lock)
547
  def ComputeDRBDMap(self):
548
    """Compute the used DRBD minor/nodes.
549

550
    This is just a wrapper over L{_UnlockedComputeDRBDMap}.
551

552
    @return: dictionary of node_name: dict of minor: instance_name;
553
        the returned dict will have all the nodes in it (even if with
554
        an empty list).
555

556
    """
557
    d_map, duplicates = self._UnlockedComputeDRBDMap()
558
    if duplicates:
559
      raise errors.ConfigurationError("Duplicate DRBD ports detected: %s" %
560
                                      str(duplicates))
561
    return d_map
562

    
563
  @locking.ssynchronized(_config_lock)
564
  def AllocateDRBDMinor(self, nodes, instance):
565
    """Allocate a drbd minor.
566

567
    The free minor will be automatically computed from the existing
568
    devices. A node can be given multiple times in order to allocate
569
    multiple minors. The result is the list of minors, in the same
570
    order as the passed nodes.
571

572
    @type instance: string
573
    @param instance: the instance for which we allocate minors
574

575
    """
576
    assert isinstance(instance, basestring), \
577
           "Invalid argument '%s' passed to AllocateDRBDMinor" % instance
578

    
579
    d_map, duplicates = self._UnlockedComputeDRBDMap()
580
    if duplicates:
581
      raise errors.ConfigurationError("Duplicate DRBD ports detected: %s" %
582
                                      str(duplicates))
583
    result = []
584
    for nname in nodes:
585
      ndata = d_map[nname]
586
      if not ndata:
587
        # no minors used, we can start at 0
588
        result.append(0)
589
        ndata[0] = instance
590
        self._temporary_drbds[(nname, 0)] = instance
591
        continue
592
      keys = ndata.keys()
593
      keys.sort()
594
      ffree = utils.FirstFree(keys)
595
      if ffree is None:
596
        # return the next minor
597
        # TODO: implement high-limit check
598
        minor = keys[-1] + 1
599
      else:
600
        minor = ffree
601
      # double-check minor against current instances
602
      assert minor not in d_map[nname], \
603
             ("Attempt to reuse allocated DRBD minor %d on node %s,"
604
              " already allocated to instance %s" %
605
              (minor, nname, d_map[nname][minor]))
606
      ndata[minor] = instance
607
      # double-check minor against reservation
608
      r_key = (nname, minor)
609
      assert r_key not in self._temporary_drbds, \
610
             ("Attempt to reuse reserved DRBD minor %d on node %s,"
611
              " reserved for instance %s" %
612
              (minor, nname, self._temporary_drbds[r_key]))
613
      self._temporary_drbds[r_key] = instance
614
      result.append(minor)
615
    logging.debug("Request to allocate drbd minors, input: %s, returning %s",
616
                  nodes, result)
617
    return result
618

    
619
  def _UnlockedReleaseDRBDMinors(self, instance):
620
    """Release temporary drbd minors allocated for a given instance.
621

622
    @type instance: string
623
    @param instance: the instance for which temporary minors should be
624
                     released
625

626
    """
627
    assert isinstance(instance, basestring), \
628
           "Invalid argument passed to ReleaseDRBDMinors"
629
    for key, name in self._temporary_drbds.items():
630
      if name == instance:
631
        del self._temporary_drbds[key]
632

    
633
  @locking.ssynchronized(_config_lock)
634
  def ReleaseDRBDMinors(self, instance):
635
    """Release temporary drbd minors allocated for a given instance.
636

637
    This should be called on the error paths, on the success paths
638
    it's automatically called by the ConfigWriter add and update
639
    functions.
640

641
    This function is just a wrapper over L{_UnlockedReleaseDRBDMinors}.
642

643
    @type instance: string
644
    @param instance: the instance for which temporary minors should be
645
                     released
646

647
    """
648
    self._UnlockedReleaseDRBDMinors(instance)
649

    
650
  @locking.ssynchronized(_config_lock, shared=1)
651
  def GetConfigVersion(self):
652
    """Get the configuration version.
653

654
    @return: Config version
655

656
    """
657
    return self._config_data.version
658

    
659
  @locking.ssynchronized(_config_lock, shared=1)
660
  def GetClusterName(self):
661
    """Get cluster name.
662

663
    @return: Cluster name
664

665
    """
666
    return self._config_data.cluster.cluster_name
667

    
668
  @locking.ssynchronized(_config_lock, shared=1)
669
  def GetMasterNode(self):
670
    """Get the hostname of the master node for this cluster.
671

672
    @return: Master hostname
673

674
    """
675
    return self._config_data.cluster.master_node
676

    
677
  @locking.ssynchronized(_config_lock, shared=1)
678
  def GetMasterIP(self):
679
    """Get the IP of the master node for this cluster.
680

681
    @return: Master IP
682

683
    """
684
    return self._config_data.cluster.master_ip
685

    
686
  @locking.ssynchronized(_config_lock, shared=1)
687
  def GetMasterNetdev(self):
688
    """Get the master network device for this cluster.
689

690
    """
691
    return self._config_data.cluster.master_netdev
692

    
693
  @locking.ssynchronized(_config_lock, shared=1)
694
  def GetFileStorageDir(self):
695
    """Get the file storage dir for this cluster.
696

697
    """
698
    return self._config_data.cluster.file_storage_dir
699

    
700
  @locking.ssynchronized(_config_lock, shared=1)
701
  def GetHypervisorType(self):
702
    """Get the hypervisor type for this cluster.
703

704
    """
705
    return self._config_data.cluster.enabled_hypervisors[0]
706

    
707
  @locking.ssynchronized(_config_lock, shared=1)
708
  def GetHostKey(self):
709
    """Return the rsa hostkey from the config.
710

711
    @rtype: string
712
    @return: the rsa hostkey
713

714
    """
715
    return self._config_data.cluster.rsahostkeypub
716

    
717
  @locking.ssynchronized(_config_lock)
718
  def AddInstance(self, instance):
719
    """Add an instance to the config.
720

721
    This should be used after creating a new instance.
722

723
    @type instance: L{objects.Instance}
724
    @param instance: the instance object
725

726
    """
727
    if not isinstance(instance, objects.Instance):
728
      raise errors.ProgrammerError("Invalid type passed to AddInstance")
729

    
730
    if instance.disk_template != constants.DT_DISKLESS:
731
      all_lvs = instance.MapLVsByNode()
732
      logging.info("Instance '%s' DISK_LAYOUT: %s", instance.name, all_lvs)
733

    
734
    all_macs = self._AllMACs()
735
    for nic in instance.nics:
736
      if nic.mac in all_macs:
737
        raise errors.ConfigurationError("Cannot add instance %s:"
738
                                        " MAC address '%s' already in use." %
739
                                        (instance.name, nic.mac))
740

    
741
    self._EnsureUUID(instance)
742

    
743
    instance.serial_no = 1
744
    instance.ctime = instance.mtime = time.time()
745
    self._config_data.instances[instance.name] = instance
746
    self._config_data.cluster.serial_no += 1
747
    self._UnlockedReleaseDRBDMinors(instance.name)
748
    for nic in instance.nics:
749
      self._temporary_macs.discard(nic.mac)
750
    self._WriteConfig()
751

    
752
  def _EnsureUUID(self, item):
753
    """Ensures a given object has a valid UUID.
754

755
    @param item: the instance or node to be checked
756

757
    """
758
    if not item.uuid:
759
      item.uuid = self._GenerateUniqueID()
760
    elif item.uuid in self._AllIDs(temporary=True):
761
      raise errors.ConfigurationError("Cannot add '%s': UUID already in use" %
762
                                      (item.name, item.uuid))
763

    
764
  def _SetInstanceStatus(self, instance_name, status):
765
    """Set the instance's status to a given value.
766

767
    """
768
    assert isinstance(status, bool), \
769
           "Invalid status '%s' passed to SetInstanceStatus" % (status,)
770

    
771
    if instance_name not in self._config_data.instances:
772
      raise errors.ConfigurationError("Unknown instance '%s'" %
773
                                      instance_name)
774
    instance = self._config_data.instances[instance_name]
775
    if instance.admin_up != status:
776
      instance.admin_up = status
777
      instance.serial_no += 1
778
      instance.mtime = time.time()
779
      self._WriteConfig()
780

    
781
  @locking.ssynchronized(_config_lock)
782
  def MarkInstanceUp(self, instance_name):
783
    """Mark the instance status to up in the config.
784

785
    """
786
    self._SetInstanceStatus(instance_name, True)
787

    
788
  @locking.ssynchronized(_config_lock)
789
  def RemoveInstance(self, instance_name):
790
    """Remove the instance from the configuration.
791

792
    """
793
    if instance_name not in self._config_data.instances:
794
      raise errors.ConfigurationError("Unknown instance '%s'" % instance_name)
795
    del self._config_data.instances[instance_name]
796
    self._config_data.cluster.serial_no += 1
797
    self._WriteConfig()
798

    
799
  @locking.ssynchronized(_config_lock)
800
  def RenameInstance(self, old_name, new_name):
801
    """Rename an instance.
802

803
    This needs to be done in ConfigWriter and not by RemoveInstance
804
    combined with AddInstance as only we can guarantee an atomic
805
    rename.
806

807
    """
808
    if old_name not in self._config_data.instances:
809
      raise errors.ConfigurationError("Unknown instance '%s'" % old_name)
810
    inst = self._config_data.instances[old_name]
811
    del self._config_data.instances[old_name]
812
    inst.name = new_name
813

    
814
    for disk in inst.disks:
815
      if disk.dev_type == constants.LD_FILE:
816
        # rename the file paths in logical and physical id
817
        file_storage_dir = os.path.dirname(os.path.dirname(disk.logical_id[1]))
818
        disk.physical_id = disk.logical_id = (disk.logical_id[0],
819
                                              os.path.join(file_storage_dir,
820
                                                           inst.name,
821
                                                           disk.iv_name))
822

    
823
    self._config_data.instances[inst.name] = inst
824
    self._WriteConfig()
825

    
826
  @locking.ssynchronized(_config_lock)
827
  def MarkInstanceDown(self, instance_name):
828
    """Mark the status of an instance to down in the configuration.
829

830
    """
831
    self._SetInstanceStatus(instance_name, False)
832

    
833
  def _UnlockedGetInstanceList(self):
834
    """Get the list of instances.
835

836
    This function is for internal use, when the config lock is already held.
837

838
    """
839
    return self._config_data.instances.keys()
840

    
841
  @locking.ssynchronized(_config_lock, shared=1)
842
  def GetInstanceList(self):
843
    """Get the list of instances.
844

845
    @return: array of instances, ex. ['instance2.example.com',
846
        'instance1.example.com']
847

848
    """
849
    return self._UnlockedGetInstanceList()
850

    
851
  @locking.ssynchronized(_config_lock, shared=1)
852
  def ExpandInstanceName(self, short_name):
853
    """Attempt to expand an incomplete instance name.
854

855
    """
856
    return utils.MatchNameComponent(short_name,
857
                                    self._config_data.instances.keys())
858

    
859
  def _UnlockedGetInstanceInfo(self, instance_name):
860
    """Returns information about an instance.
861

862
    This function is for internal use, when the config lock is already held.
863

864
    """
865
    if instance_name not in self._config_data.instances:
866
      return None
867

    
868
    return self._config_data.instances[instance_name]
869

    
870
  @locking.ssynchronized(_config_lock, shared=1)
871
  def GetInstanceInfo(self, instance_name):
872
    """Returns information about an instance.
873

874
    It takes the information from the configuration file. Other information of
875
    an instance are taken from the live systems.
876

877
    @param instance_name: name of the instance, e.g.
878
        I{instance1.example.com}
879

880
    @rtype: L{objects.Instance}
881
    @return: the instance object
882

883
    """
884
    return self._UnlockedGetInstanceInfo(instance_name)
885

    
886
  @locking.ssynchronized(_config_lock, shared=1)
887
  def GetAllInstancesInfo(self):
888
    """Get the configuration of all instances.
889

890
    @rtype: dict
891
    @return: dict of (instance, instance_info), where instance_info is what
892
              would GetInstanceInfo return for the node
893

894
    """
895
    my_dict = dict([(instance, self._UnlockedGetInstanceInfo(instance))
896
                    for instance in self._UnlockedGetInstanceList()])
897
    return my_dict
898

    
899
  @locking.ssynchronized(_config_lock)
900
  def AddNode(self, node):
901
    """Add a node to the configuration.
902

903
    @type node: L{objects.Node}
904
    @param node: a Node instance
905

906
    """
907
    logging.info("Adding node %s to configuration" % node.name)
908

    
909
    self._EnsureUUID(node)
910

    
911
    node.serial_no = 1
912
    node.ctime = node.mtime = time.time()
913
    self._config_data.nodes[node.name] = node
914
    self._config_data.cluster.serial_no += 1
915
    self._WriteConfig()
916

    
917
  @locking.ssynchronized(_config_lock)
918
  def RemoveNode(self, node_name):
919
    """Remove a node from the configuration.
920

921
    """
922
    logging.info("Removing node %s from configuration" % node_name)
923

    
924
    if node_name not in self._config_data.nodes:
925
      raise errors.ConfigurationError("Unknown node '%s'" % node_name)
926

    
927
    del self._config_data.nodes[node_name]
928
    self._config_data.cluster.serial_no += 1
929
    self._WriteConfig()
930

    
931
  @locking.ssynchronized(_config_lock, shared=1)
932
  def ExpandNodeName(self, short_name):
933
    """Attempt to expand an incomplete instance name.
934

935
    """
936
    return utils.MatchNameComponent(short_name,
937
                                    self._config_data.nodes.keys())
938

    
939
  def _UnlockedGetNodeInfo(self, node_name):
940
    """Get the configuration of a node, as stored in the config.
941

942
    This function is for internal use, when the config lock is already
943
    held.
944

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

947
    @rtype: L{objects.Node}
948
    @return: the node object
949

950
    """
951
    if node_name not in self._config_data.nodes:
952
      return None
953

    
954
    return self._config_data.nodes[node_name]
955

    
956

    
957
  @locking.ssynchronized(_config_lock, shared=1)
958
  def GetNodeInfo(self, node_name):
959
    """Get the configuration of a node, as stored in the config.
960

961
    This is just a locked wrapper over L{_UnlockedGetNodeInfo}.
962

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

965
    @rtype: L{objects.Node}
966
    @return: the node object
967

968
    """
969
    return self._UnlockedGetNodeInfo(node_name)
970

    
971
  def _UnlockedGetNodeList(self):
972
    """Return the list of nodes which are in the configuration.
973

974
    This function is for internal use, when the config lock is already
975
    held.
976

977
    @rtype: list
978

979
    """
980
    return self._config_data.nodes.keys()
981

    
982

    
983
  @locking.ssynchronized(_config_lock, shared=1)
984
  def GetNodeList(self):
985
    """Return the list of nodes which are in the configuration.
986

987
    """
988
    return self._UnlockedGetNodeList()
989

    
990
  @locking.ssynchronized(_config_lock, shared=1)
991
  def GetOnlineNodeList(self):
992
    """Return the list of nodes which are online.
993

994
    """
995
    all_nodes = [self._UnlockedGetNodeInfo(node)
996
                 for node in self._UnlockedGetNodeList()]
997
    return [node.name for node in all_nodes if not node.offline]
998

    
999
  @locking.ssynchronized(_config_lock, shared=1)
1000
  def GetAllNodesInfo(self):
1001
    """Get the configuration of all nodes.
1002

1003
    @rtype: dict
1004
    @return: dict of (node, node_info), where node_info is what
1005
              would GetNodeInfo return for the node
1006

1007
    """
1008
    my_dict = dict([(node, self._UnlockedGetNodeInfo(node))
1009
                    for node in self._UnlockedGetNodeList()])
1010
    return my_dict
1011

    
1012
  def _UnlockedGetMasterCandidateStats(self, exceptions=None):
1013
    """Get the number of current and maximum desired and possible candidates.
1014

1015
    @type exceptions: list
1016
    @param exceptions: if passed, list of nodes that should be ignored
1017
    @rtype: tuple
1018
    @return: tuple of (current, desired and possible, possible)
1019

1020
    """
1021
    mc_now = mc_should = mc_max = 0
1022
    for node in self._config_data.nodes.values():
1023
      if exceptions and node.name in exceptions:
1024
        continue
1025
      if not (node.offline or node.drained):
1026
        mc_max += 1
1027
      if node.master_candidate:
1028
        mc_now += 1
1029
    mc_should = min(mc_max, self._config_data.cluster.candidate_pool_size)
1030
    return (mc_now, mc_should, mc_max)
1031

    
1032
  @locking.ssynchronized(_config_lock, shared=1)
1033
  def GetMasterCandidateStats(self, exceptions=None):
1034
    """Get the number of current and maximum possible candidates.
1035

1036
    This is just a wrapper over L{_UnlockedGetMasterCandidateStats}.
1037

1038
    @type exceptions: list
1039
    @param exceptions: if passed, list of nodes that should be ignored
1040
    @rtype: tuple
1041
    @return: tuple of (current, max)
1042

1043
    """
1044
    return self._UnlockedGetMasterCandidateStats(exceptions)
1045

    
1046
  @locking.ssynchronized(_config_lock)
1047
  def MaintainCandidatePool(self, exceptions):
1048
    """Try to grow the candidate pool to the desired size.
1049

1050
    @type exceptions: list
1051
    @param exceptions: if passed, list of nodes that should be ignored
1052
    @rtype: list
1053
    @return: list with the adjusted nodes (L{objects.Node} instances)
1054

1055
    """
1056
    mc_now, mc_max, _ = self._UnlockedGetMasterCandidateStats(exceptions)
1057
    mod_list = []
1058
    if mc_now < mc_max:
1059
      node_list = self._config_data.nodes.keys()
1060
      random.shuffle(node_list)
1061
      for name in node_list:
1062
        if mc_now >= mc_max:
1063
          break
1064
        node = self._config_data.nodes[name]
1065
        if (node.master_candidate or node.offline or node.drained or
1066
            node.name in exceptions):
1067
          continue
1068
        mod_list.append(node)
1069
        node.master_candidate = True
1070
        node.serial_no += 1
1071
        mc_now += 1
1072
      if mc_now != mc_max:
1073
        # this should not happen
1074
        logging.warning("Warning: MaintainCandidatePool didn't manage to"
1075
                        " fill the candidate pool (%d/%d)", mc_now, mc_max)
1076
      if mod_list:
1077
        self._config_data.cluster.serial_no += 1
1078
        self._WriteConfig()
1079

    
1080
    return mod_list
1081

    
1082
  def _BumpSerialNo(self):
1083
    """Bump up the serial number of the config.
1084

1085
    """
1086
    self._config_data.serial_no += 1
1087
    self._config_data.mtime = time.time()
1088

    
1089
  def _AllUUIDObjects(self):
1090
    """Returns all objects with uuid attributes.
1091

1092
    """
1093
    return (self._config_data.instances.values() +
1094
            self._config_data.nodes.values() +
1095
            [self._config_data.cluster])
1096

    
1097
  def _OpenConfig(self):
1098
    """Read the config data from disk.
1099

1100
    """
1101
    raw_data = utils.ReadFile(self._cfg_file)
1102

    
1103
    try:
1104
      data = objects.ConfigData.FromDict(serializer.Load(raw_data))
1105
    except Exception, err:
1106
      raise errors.ConfigurationError(err)
1107

    
1108
    # Make sure the configuration has the right version
1109
    _ValidateConfig(data)
1110

    
1111
    if (not hasattr(data, 'cluster') or
1112
        not hasattr(data.cluster, 'rsahostkeypub')):
1113
      raise errors.ConfigurationError("Incomplete configuration"
1114
                                      " (missing cluster.rsahostkeypub)")
1115

    
1116
    # Upgrade configuration if needed
1117
    data.UpgradeConfig()
1118

    
1119
    self._config_data = data
1120
    # reset the last serial as -1 so that the next write will cause
1121
    # ssconf update
1122
    self._last_cluster_serial = -1
1123

    
1124
    # And finally run our (custom) config upgrade sequence
1125
    self._UpgradeConfig()
1126

    
1127
  def _UpgradeConfig(self):
1128
    """Run upgrade steps that cannot be done purely in the objects.
1129

1130
    This is because some data elements need uniqueness across the
1131
    whole configuration, etc.
1132

1133
    @warning: this function will call L{_WriteConfig()}, so it needs
1134
        to either be called with the lock held or from a safe place
1135
        (the constructor)
1136

1137
    """
1138
    modified = False
1139
    for item in self._AllUUIDObjects():
1140
      if item.uuid is None:
1141
        item.uuid = self._GenerateUniqueID()
1142
        modified = True
1143
    if modified:
1144
      self._WriteConfig()
1145

    
1146
  def _DistributeConfig(self):
1147
    """Distribute the configuration to the other nodes.
1148

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

1152
    """
1153
    if self._offline:
1154
      return True
1155
    bad = False
1156

    
1157
    node_list = []
1158
    addr_list = []
1159
    myhostname = self._my_hostname
1160
    # we can skip checking whether _UnlockedGetNodeInfo returns None
1161
    # since the node list comes from _UnlocketGetNodeList, and we are
1162
    # called with the lock held, so no modifications should take place
1163
    # in between
1164
    for node_name in self._UnlockedGetNodeList():
1165
      if node_name == myhostname:
1166
        continue
1167
      node_info = self._UnlockedGetNodeInfo(node_name)
1168
      if not node_info.master_candidate:
1169
        continue
1170
      node_list.append(node_info.name)
1171
      addr_list.append(node_info.primary_ip)
1172

    
1173
    result = rpc.RpcRunner.call_upload_file(node_list, self._cfg_file,
1174
                                            address_list=addr_list)
1175
    for to_node, to_result in result.items():
1176
      msg = to_result.fail_msg
1177
      if msg:
1178
        msg = ("Copy of file %s to node %s failed: %s" %
1179
               (self._cfg_file, to_node, msg))
1180
        logging.error(msg)
1181
        bad = True
1182
    return not bad
1183

    
1184
  def _WriteConfig(self, destination=None):
1185
    """Write the configuration data to persistent storage.
1186

1187
    """
1188
    # first, cleanup the _temporary_ids set, if an ID is now in the
1189
    # other objects it should be discarded to prevent unbounded growth
1190
    # of that structure
1191
    self._CleanupTemporaryIDs()
1192
    config_errors = self._UnlockedVerifyConfig()
1193
    if config_errors:
1194
      raise errors.ConfigurationError("Configuration data is not"
1195
                                      " consistent: %s" %
1196
                                      (", ".join(config_errors)))
1197
    if destination is None:
1198
      destination = self._cfg_file
1199
    self._BumpSerialNo()
1200
    txt = serializer.Dump(self._config_data.ToDict())
1201

    
1202
    utils.WriteFile(destination, data=txt)
1203

    
1204
    self.write_count += 1
1205

    
1206
    # and redistribute the config file to master candidates
1207
    self._DistributeConfig()
1208

    
1209
    # Write ssconf files on all nodes (including locally)
1210
    if self._last_cluster_serial < self._config_data.cluster.serial_no:
1211
      if not self._offline:
1212
        result = rpc.RpcRunner.call_write_ssconf_files(\
1213
          self._UnlockedGetNodeList(),
1214
          self._UnlockedGetSsconfValues())
1215
        for nname, nresu in result.items():
1216
          msg = nresu.fail_msg
1217
          if msg:
1218
            logging.warning("Error while uploading ssconf files to"
1219
                            " node %s: %s", nname, msg)
1220
      self._last_cluster_serial = self._config_data.cluster.serial_no
1221

    
1222
  def _UnlockedGetSsconfValues(self):
1223
    """Return the values needed by ssconf.
1224

1225
    @rtype: dict
1226
    @return: a dictionary with keys the ssconf names and values their
1227
        associated value
1228

1229
    """
1230
    fn = "\n".join
1231
    instance_names = utils.NiceSort(self._UnlockedGetInstanceList())
1232
    node_names = utils.NiceSort(self._UnlockedGetNodeList())
1233
    node_info = [self._UnlockedGetNodeInfo(name) for name in node_names]
1234
    node_pri_ips = ["%s %s" % (ninfo.name, ninfo.primary_ip)
1235
                    for ninfo in node_info]
1236
    node_snd_ips = ["%s %s" % (ninfo.name, ninfo.secondary_ip)
1237
                    for ninfo in node_info]
1238

    
1239
    instance_data = fn(instance_names)
1240
    off_data = fn(node.name for node in node_info if node.offline)
1241
    on_data = fn(node.name for node in node_info if not node.offline)
1242
    mc_data = fn(node.name for node in node_info if node.master_candidate)
1243
    mc_ips_data = fn(node.primary_ip for node in node_info
1244
                     if node.master_candidate)
1245
    node_data = fn(node_names)
1246
    node_pri_ips_data = fn(node_pri_ips)
1247
    node_snd_ips_data = fn(node_snd_ips)
1248

    
1249
    cluster = self._config_data.cluster
1250
    cluster_tags = fn(cluster.GetTags())
1251
    return {
1252
      constants.SS_CLUSTER_NAME: cluster.cluster_name,
1253
      constants.SS_CLUSTER_TAGS: cluster_tags,
1254
      constants.SS_FILE_STORAGE_DIR: cluster.file_storage_dir,
1255
      constants.SS_MASTER_CANDIDATES: mc_data,
1256
      constants.SS_MASTER_CANDIDATES_IPS: mc_ips_data,
1257
      constants.SS_MASTER_IP: cluster.master_ip,
1258
      constants.SS_MASTER_NETDEV: cluster.master_netdev,
1259
      constants.SS_MASTER_NODE: cluster.master_node,
1260
      constants.SS_NODE_LIST: node_data,
1261
      constants.SS_NODE_PRIMARY_IPS: node_pri_ips_data,
1262
      constants.SS_NODE_SECONDARY_IPS: node_snd_ips_data,
1263
      constants.SS_OFFLINE_NODES: off_data,
1264
      constants.SS_ONLINE_NODES: on_data,
1265
      constants.SS_INSTANCE_LIST: instance_data,
1266
      constants.SS_RELEASE_VERSION: constants.RELEASE_VERSION,
1267
      }
1268

    
1269
  @locking.ssynchronized(_config_lock, shared=1)
1270
  def GetVGName(self):
1271
    """Return the volume group name.
1272

1273
    """
1274
    return self._config_data.cluster.volume_group_name
1275

    
1276
  @locking.ssynchronized(_config_lock)
1277
  def SetVGName(self, vg_name):
1278
    """Set the volume group name.
1279

1280
    """
1281
    self._config_data.cluster.volume_group_name = vg_name
1282
    self._config_data.cluster.serial_no += 1
1283
    self._WriteConfig()
1284

    
1285
  @locking.ssynchronized(_config_lock, shared=1)
1286
  def GetMACPrefix(self):
1287
    """Return the mac prefix.
1288

1289
    """
1290
    return self._config_data.cluster.mac_prefix
1291

    
1292
  @locking.ssynchronized(_config_lock, shared=1)
1293
  def GetClusterInfo(self):
1294
    """Returns information about the cluster
1295

1296
    @rtype: L{objects.Cluster}
1297
    @return: the cluster object
1298

1299
    """
1300
    return self._config_data.cluster
1301

    
1302
  @locking.ssynchronized(_config_lock)
1303
  def Update(self, target):
1304
    """Notify function to be called after updates.
1305

1306
    This function must be called when an object (as returned by
1307
    GetInstanceInfo, GetNodeInfo, GetCluster) has been updated and the
1308
    caller wants the modifications saved to the backing store. Note
1309
    that all modified objects will be saved, but the target argument
1310
    is the one the caller wants to ensure that it's saved.
1311

1312
    @param target: an instance of either L{objects.Cluster},
1313
        L{objects.Node} or L{objects.Instance} which is existing in
1314
        the cluster
1315

1316
    """
1317
    if self._config_data is None:
1318
      raise errors.ProgrammerError("Configuration file not read,"
1319
                                   " cannot save.")
1320
    update_serial = False
1321
    if isinstance(target, objects.Cluster):
1322
      test = target == self._config_data.cluster
1323
    elif isinstance(target, objects.Node):
1324
      test = target in self._config_data.nodes.values()
1325
      update_serial = True
1326
    elif isinstance(target, objects.Instance):
1327
      test = target in self._config_data.instances.values()
1328
    else:
1329
      raise errors.ProgrammerError("Invalid object type (%s) passed to"
1330
                                   " ConfigWriter.Update" % type(target))
1331
    if not test:
1332
      raise errors.ConfigurationError("Configuration updated since object"
1333
                                      " has been read or unknown object")
1334
    target.serial_no += 1
1335
    target.mtime = now = time.time()
1336

    
1337
    if update_serial:
1338
      # for node updates, we need to increase the cluster serial too
1339
      self._config_data.cluster.serial_no += 1
1340
      self._config_data.cluster.mtime = now
1341

    
1342
    if isinstance(target, objects.Instance):
1343
      self._UnlockedReleaseDRBDMinors(target.name)
1344
      for nic in target.nics:
1345
        self._temporary_macs.discard(nic.mac)
1346

    
1347
    self._WriteConfig()