Statistics
| Branch: | Tag: | Revision:

root / lib / config.py @ e6345c35

History | View | Annotate | Download (40.9 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 _AllLVs(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
  def _AllIDs(self, include_temporary):
163
    """Compute the list of all UUIDs and names we have.
164

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

170
    """
171
    existing = set()
172
    if include_temporary:
173
      existing.update(self._temporary_ids)
174
    existing.update(self._AllLVs())
175
    existing.update(self._config_data.instances.keys())
176
    existing.update(self._config_data.nodes.keys())
177
    return existing
178

    
179
  @locking.ssynchronized(_config_lock, shared=1)
180
  def GenerateUniqueID(self, exceptions=None):
181
    """Generate an unique disk name.
182

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

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

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

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

    
209
  def _CleanupTemporaryIDs(self):
210
    """Cleanups the _temporary_ids structure.
211

212
    """
213
    existing = self._AllIDs(include_temporary=False)
214
    self._temporary_ids = self._temporary_ids - existing
215

    
216
  def _AllMACs(self):
217
    """Return all MACs present in the config.
218

219
    @rtype: list
220
    @return: the list of all MACs
221

222
    """
223
    result = []
224
    for instance in self._config_data.instances.values():
225
      for nic in instance.nics:
226
        result.append(nic.mac)
227

    
228
    return result
229

    
230
  def _AllDRBDSecrets(self):
231
    """Return all DRBD secrets present in the config.
232

233
    @rtype: list
234
    @return: the list of all DRBD secrets
235

236
    """
237
    def helper(disk, result):
238
      """Recursively gather secrets from this disk."""
239
      if disk.dev_type == constants.DT_DRBD8:
240
        result.append(disk.logical_id[5])
241
      if disk.children:
242
        for child in disk.children:
243
          helper(child, result)
244

    
245
    result = []
246
    for instance in self._config_data.instances.values():
247
      for disk in instance.disks:
248
        helper(disk, result)
249

    
250
    return result
251

    
252
  def _CheckDiskIDs(self, disk, l_ids, p_ids):
253
    """Compute duplicate disk IDs
254

255
    @type disk: L{objects.Disk}
256
    @param disk: the disk at which to start searching
257
    @type l_ids: list
258
    @param l_ids: list of current logical ids
259
    @type p_ids: list
260
    @param p_ids: list of current physical ids
261
    @rtype: list
262
    @return: a list of error messages
263

264
    """
265
    result = []
266
    if disk.logical_id is not None:
267
      if disk.logical_id in l_ids:
268
        result.append("duplicate logical id %s" % str(disk.logical_id))
269
      else:
270
        l_ids.append(disk.logical_id)
271
    if disk.physical_id is not None:
272
      if disk.physical_id in p_ids:
273
        result.append("duplicate physical id %s" % str(disk.physical_id))
274
      else:
275
        p_ids.append(disk.physical_id)
276

    
277
    if disk.children:
278
      for child in disk.children:
279
        result.extend(self._CheckDiskIDs(child, l_ids, p_ids))
280
    return result
281

    
282
  def _UnlockedVerifyConfig(self):
283
    """Verify function.
284

285
    @rtype: list
286
    @return: a list of error messages; a non-empty list signifies
287
        configuration errors
288

289
    """
290
    result = []
291
    seen_macs = []
292
    ports = {}
293
    data = self._config_data
294
    seen_lids = []
295
    seen_pids = []
296

    
297
    # global cluster checks
298
    if not data.cluster.enabled_hypervisors:
299
      result.append("enabled hypervisors list doesn't have any entries")
300
    invalid_hvs = set(data.cluster.enabled_hypervisors) - constants.HYPER_TYPES
301
    if invalid_hvs:
302
      result.append("enabled hypervisors contains invalid entries: %s" %
303
                    invalid_hvs)
304

    
305
    if data.cluster.master_node not in data.nodes:
306
      result.append("cluster has invalid primary node '%s'" %
307
                    data.cluster.master_node)
308

    
309
    # per-instance checks
310
    for instance_name in data.instances:
311
      instance = data.instances[instance_name]
312
      if instance.primary_node not in data.nodes:
313
        result.append("instance '%s' has invalid primary node '%s'" %
314
                      (instance_name, instance.primary_node))
315
      for snode in instance.secondary_nodes:
316
        if snode not in data.nodes:
317
          result.append("instance '%s' has invalid secondary node '%s'" %
318
                        (instance_name, snode))
319
      for idx, nic in enumerate(instance.nics):
320
        if nic.mac in seen_macs:
321
          result.append("instance '%s' has NIC %d mac %s duplicate" %
322
                        (instance_name, idx, nic.mac))
323
        else:
324
          seen_macs.append(nic.mac)
325

    
326
      # gather the drbd ports for duplicate checks
327
      for dsk in instance.disks:
328
        if dsk.dev_type in constants.LDS_DRBD:
329
          tcp_port = dsk.logical_id[2]
330
          if tcp_port not in ports:
331
            ports[tcp_port] = []
332
          ports[tcp_port].append((instance.name, "drbd disk %s" % dsk.iv_name))
333
      # gather network port reservation
334
      net_port = getattr(instance, "network_port", None)
335
      if net_port is not None:
336
        if net_port not in ports:
337
          ports[net_port] = []
338
        ports[net_port].append((instance.name, "network port"))
339

    
340
      # instance disk verify
341
      for idx, disk in enumerate(instance.disks):
342
        result.extend(["instance '%s' disk %d error: %s" %
343
                       (instance.name, idx, msg) for msg in disk.Verify()])
344
        result.extend(self._CheckDiskIDs(disk, seen_lids, seen_pids))
345

    
346
    # cluster-wide pool of free ports
347
    for free_port in data.cluster.tcpudp_port_pool:
348
      if free_port not in ports:
349
        ports[free_port] = []
350
      ports[free_port].append(("cluster", "port marked as free"))
351

    
352
    # compute tcp/udp duplicate ports
353
    keys = ports.keys()
354
    keys.sort()
355
    for pnum in keys:
356
      pdata = ports[pnum]
357
      if len(pdata) > 1:
358
        txt = ", ".join(["%s/%s" % val for val in pdata])
359
        result.append("tcp/udp port %s has duplicates: %s" % (pnum, txt))
360

    
361
    # highest used tcp port check
362
    if keys:
363
      if keys[-1] > data.cluster.highest_used_port:
364
        result.append("Highest used port mismatch, saved %s, computed %s" %
365
                      (data.cluster.highest_used_port, keys[-1]))
366

    
367
    if not data.nodes[data.cluster.master_node].master_candidate:
368
      result.append("Master node is not a master candidate")
369

    
370
    # master candidate checks
371
    mc_now, mc_max = self._UnlockedGetMasterCandidateStats()
372
    if mc_now < mc_max:
373
      result.append("Not enough master candidates: actual %d, target %d" %
374
                    (mc_now, mc_max))
375

    
376
    # node checks
377
    for node in data.nodes.values():
378
      if [node.master_candidate, node.drained, node.offline].count(True) > 1:
379
        result.append("Node %s state is invalid: master_candidate=%s,"
380
                      " drain=%s, offline=%s" %
381
                      (node.name, node.master_candidate, node.drain,
382
                       node.offline))
383

    
384
    # drbd minors check
385
    d_map, duplicates = self._UnlockedComputeDRBDMap()
386
    for node, minor, instance_a, instance_b in duplicates:
387
      result.append("DRBD minor %d on node %s is assigned twice to instances"
388
                    " %s and %s" % (minor, node, instance_a, instance_b))
389

    
390
    return result
391

    
392
  @locking.ssynchronized(_config_lock, shared=1)
393
  def VerifyConfig(self):
394
    """Verify function.
395

396
    This is just a wrapper over L{_UnlockedVerifyConfig}.
397

398
    @rtype: list
399
    @return: a list of error messages; a non-empty list signifies
400
        configuration errors
401

402
    """
403
    return self._UnlockedVerifyConfig()
404

    
405
  def _UnlockedSetDiskID(self, disk, node_name):
406
    """Convert the unique ID to the ID needed on the target nodes.
407

408
    This is used only for drbd, which needs ip/port configuration.
409

410
    The routine descends down and updates its children also, because
411
    this helps when the only the top device is passed to the remote
412
    node.
413

414
    This function is for internal use, when the config lock is already held.
415

416
    """
417
    if disk.children:
418
      for child in disk.children:
419
        self._UnlockedSetDiskID(child, node_name)
420

    
421
    if disk.logical_id is None and disk.physical_id is not None:
422
      return
423
    if disk.dev_type == constants.LD_DRBD8:
424
      pnode, snode, port, pminor, sminor, secret = disk.logical_id
425
      if node_name not in (pnode, snode):
426
        raise errors.ConfigurationError("DRBD device not knowing node %s" %
427
                                        node_name)
428
      pnode_info = self._UnlockedGetNodeInfo(pnode)
429
      snode_info = self._UnlockedGetNodeInfo(snode)
430
      if pnode_info is None or snode_info is None:
431
        raise errors.ConfigurationError("Can't find primary or secondary node"
432
                                        " for %s" % str(disk))
433
      p_data = (pnode_info.secondary_ip, port)
434
      s_data = (snode_info.secondary_ip, port)
435
      if pnode == node_name:
436
        disk.physical_id = p_data + s_data + (pminor, secret)
437
      else: # it must be secondary, we tested above
438
        disk.physical_id = s_data + p_data + (sminor, secret)
439
    else:
440
      disk.physical_id = disk.logical_id
441
    return
442

    
443
  @locking.ssynchronized(_config_lock)
444
  def SetDiskID(self, disk, node_name):
445
    """Convert the unique ID to the ID needed on the target nodes.
446

447
    This is used only for drbd, which needs ip/port configuration.
448

449
    The routine descends down and updates its children also, because
450
    this helps when the only the top device is passed to the remote
451
    node.
452

453
    """
454
    return self._UnlockedSetDiskID(disk, node_name)
455

    
456
  @locking.ssynchronized(_config_lock)
457
  def AddTcpUdpPort(self, port):
458
    """Adds a new port to the available port pool.
459

460
    """
461
    if not isinstance(port, int):
462
      raise errors.ProgrammerError("Invalid type passed for port")
463

    
464
    self._config_data.cluster.tcpudp_port_pool.add(port)
465
    self._WriteConfig()
466

    
467
  @locking.ssynchronized(_config_lock, shared=1)
468
  def GetPortList(self):
469
    """Returns a copy of the current port list.
470

471
    """
472
    return self._config_data.cluster.tcpudp_port_pool.copy()
473

    
474
  @locking.ssynchronized(_config_lock)
475
  def AllocatePort(self):
476
    """Allocate a port.
477

478
    The port will be taken from the available port pool or from the
479
    default port range (and in this case we increase
480
    highest_used_port).
481

482
    """
483
    # If there are TCP/IP ports configured, we use them first.
484
    if self._config_data.cluster.tcpudp_port_pool:
485
      port = self._config_data.cluster.tcpudp_port_pool.pop()
486
    else:
487
      port = self._config_data.cluster.highest_used_port + 1
488
      if port >= constants.LAST_DRBD_PORT:
489
        raise errors.ConfigurationError("The highest used port is greater"
490
                                        " than %s. Aborting." %
491
                                        constants.LAST_DRBD_PORT)
492
      self._config_data.cluster.highest_used_port = port
493

    
494
    self._WriteConfig()
495
    return port
496

    
497
  def _UnlockedComputeDRBDMap(self):
498
    """Compute the used DRBD minor/nodes.
499

500
    @rtype: (dict, list)
501
    @return: dictionary of node_name: dict of minor: instance_name;
502
        the returned dict will have all the nodes in it (even if with
503
        an empty list), and a list of duplicates; if the duplicates
504
        list is not empty, the configuration is corrupted and its caller
505
        should raise an exception
506

507
    """
508
    def _AppendUsedPorts(instance_name, disk, used):
509
      duplicates = []
510
      if disk.dev_type == constants.LD_DRBD8 and len(disk.logical_id) >= 5:
511
        node_a, node_b, _, minor_a, minor_b = disk.logical_id[:5]
512
        for node, port in ((node_a, minor_a), (node_b, minor_b)):
513
          assert node in used, ("Node '%s' of instance '%s' not found"
514
                                " in node list" % (node, instance_name))
515
          if port in used[node]:
516
            duplicates.append((node, port, instance_name, used[node][port]))
517
          else:
518
            used[node][port] = instance_name
519
      if disk.children:
520
        for child in disk.children:
521
          duplicates.extend(_AppendUsedPorts(instance_name, child, used))
522
      return duplicates
523

    
524
    duplicates = []
525
    my_dict = dict((node, {}) for node in self._config_data.nodes)
526
    for instance in self._config_data.instances.itervalues():
527
      for disk in instance.disks:
528
        duplicates.extend(_AppendUsedPorts(instance.name, disk, my_dict))
529
    for (node, minor), instance in self._temporary_drbds.iteritems():
530
      if minor in my_dict[node] and my_dict[node][minor] != instance:
531
        duplicates.append((node, minor, instance, my_dict[node][minor]))
532
      else:
533
        my_dict[node][minor] = instance
534
    return my_dict, duplicates
535

    
536
  @locking.ssynchronized(_config_lock)
537
  def ComputeDRBDMap(self):
538
    """Compute the used DRBD minor/nodes.
539

540
    This is just a wrapper over L{_UnlockedComputeDRBDMap}.
541

542
    @return: dictionary of node_name: dict of minor: instance_name;
543
        the returned dict will have all the nodes in it (even if with
544
        an empty list).
545

546
    """
547
    d_map, duplicates = self._UnlockedComputeDRBDMap()
548
    if duplicates:
549
      raise errors.ConfigurationError("Duplicate DRBD ports detected: %s" %
550
                                      str(duplicates))
551
    return d_map
552

    
553
  @locking.ssynchronized(_config_lock)
554
  def AllocateDRBDMinor(self, nodes, instance):
555
    """Allocate a drbd minor.
556

557
    The free minor will be automatically computed from the existing
558
    devices. A node can be given multiple times in order to allocate
559
    multiple minors. The result is the list of minors, in the same
560
    order as the passed nodes.
561

562
    @type instance: string
563
    @param instance: the instance for which we allocate minors
564

565
    """
566
    assert isinstance(instance, basestring), \
567
           "Invalid argument '%s' passed to AllocateDRBDMinor" % instance
568

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

    
609
  def _UnlockedReleaseDRBDMinors(self, instance):
610
    """Release temporary drbd minors allocated for a given instance.
611

612
    @type instance: string
613
    @param instance: the instance for which temporary minors should be
614
                     released
615

616
    """
617
    assert isinstance(instance, basestring), \
618
           "Invalid argument passed to ReleaseDRBDMinors"
619
    for key, name in self._temporary_drbds.items():
620
      if name == instance:
621
        del self._temporary_drbds[key]
622

    
623
  @locking.ssynchronized(_config_lock)
624
  def ReleaseDRBDMinors(self, instance):
625
    """Release temporary drbd minors allocated for a given instance.
626

627
    This should be called on the error paths, on the success paths
628
    it's automatically called by the ConfigWriter add and update
629
    functions.
630

631
    This function is just a wrapper over L{_UnlockedReleaseDRBDMinors}.
632

633
    @type instance: string
634
    @param instance: the instance for which temporary minors should be
635
                     released
636

637
    """
638
    self._UnlockedReleaseDRBDMinors(instance)
639

    
640
  @locking.ssynchronized(_config_lock, shared=1)
641
  def GetConfigVersion(self):
642
    """Get the configuration version.
643

644
    @return: Config version
645

646
    """
647
    return self._config_data.version
648

    
649
  @locking.ssynchronized(_config_lock, shared=1)
650
  def GetClusterName(self):
651
    """Get cluster name.
652

653
    @return: Cluster name
654

655
    """
656
    return self._config_data.cluster.cluster_name
657

    
658
  @locking.ssynchronized(_config_lock, shared=1)
659
  def GetMasterNode(self):
660
    """Get the hostname of the master node for this cluster.
661

662
    @return: Master hostname
663

664
    """
665
    return self._config_data.cluster.master_node
666

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

671
    @return: Master IP
672

673
    """
674
    return self._config_data.cluster.master_ip
675

    
676
  @locking.ssynchronized(_config_lock, shared=1)
677
  def GetMasterNetdev(self):
678
    """Get the master network device for this cluster.
679

680
    """
681
    return self._config_data.cluster.master_netdev
682

    
683
  @locking.ssynchronized(_config_lock, shared=1)
684
  def GetFileStorageDir(self):
685
    """Get the file storage dir for this cluster.
686

687
    """
688
    return self._config_data.cluster.file_storage_dir
689

    
690
  @locking.ssynchronized(_config_lock, shared=1)
691
  def GetHypervisorType(self):
692
    """Get the hypervisor type for this cluster.
693

694
    """
695
    return self._config_data.cluster.default_hypervisor
696

    
697
  @locking.ssynchronized(_config_lock, shared=1)
698
  def GetHostKey(self):
699
    """Return the rsa hostkey from the config.
700

701
    @rtype: string
702
    @return: the rsa hostkey
703

704
    """
705
    return self._config_data.cluster.rsahostkeypub
706

    
707
  @locking.ssynchronized(_config_lock)
708
  def AddInstance(self, instance):
709
    """Add an instance to the config.
710

711
    This should be used after creating a new instance.
712

713
    @type instance: L{objects.Instance}
714
    @param instance: the instance object
715

716
    """
717
    if not isinstance(instance, objects.Instance):
718
      raise errors.ProgrammerError("Invalid type passed to AddInstance")
719

    
720
    if instance.disk_template != constants.DT_DISKLESS:
721
      all_lvs = instance.MapLVsByNode()
722
      logging.info("Instance '%s' DISK_LAYOUT: %s", instance.name, all_lvs)
723

    
724
    all_macs = self._AllMACs()
725
    for nic in instance.nics:
726
      if nic.mac in all_macs:
727
        raise errors.ConfigurationError("Cannot add instance %s:"
728
          " MAC address '%s' already in use." % (instance.name, nic.mac))
729

    
730
    instance.serial_no = 1
731
    self._config_data.instances[instance.name] = instance
732
    self._config_data.cluster.serial_no += 1
733
    self._UnlockedReleaseDRBDMinors(instance.name)
734
    for nic in instance.nics:
735
      self._temporary_macs.discard(nic.mac)
736
    self._WriteConfig()
737

    
738
  def _SetInstanceStatus(self, instance_name, status):
739
    """Set the instance's status to a given value.
740

741
    """
742
    assert isinstance(status, bool), \
743
           "Invalid status '%s' passed to SetInstanceStatus" % (status,)
744

    
745
    if instance_name not in self._config_data.instances:
746
      raise errors.ConfigurationError("Unknown instance '%s'" %
747
                                      instance_name)
748
    instance = self._config_data.instances[instance_name]
749
    if instance.admin_up != status:
750
      instance.admin_up = status
751
      instance.serial_no += 1
752
      self._WriteConfig()
753

    
754
  @locking.ssynchronized(_config_lock)
755
  def MarkInstanceUp(self, instance_name):
756
    """Mark the instance status to up in the config.
757

758
    """
759
    self._SetInstanceStatus(instance_name, True)
760

    
761
  @locking.ssynchronized(_config_lock)
762
  def RemoveInstance(self, instance_name):
763
    """Remove the instance from the configuration.
764

765
    """
766
    if instance_name not in self._config_data.instances:
767
      raise errors.ConfigurationError("Unknown instance '%s'" % instance_name)
768
    del self._config_data.instances[instance_name]
769
    self._config_data.cluster.serial_no += 1
770
    self._WriteConfig()
771

    
772
  @locking.ssynchronized(_config_lock)
773
  def RenameInstance(self, old_name, new_name):
774
    """Rename an instance.
775

776
    This needs to be done in ConfigWriter and not by RemoveInstance
777
    combined with AddInstance as only we can guarantee an atomic
778
    rename.
779

780
    """
781
    if old_name not in self._config_data.instances:
782
      raise errors.ConfigurationError("Unknown instance '%s'" % old_name)
783
    inst = self._config_data.instances[old_name]
784
    del self._config_data.instances[old_name]
785
    inst.name = new_name
786

    
787
    for disk in inst.disks:
788
      if disk.dev_type == constants.LD_FILE:
789
        # rename the file paths in logical and physical id
790
        file_storage_dir = os.path.dirname(os.path.dirname(disk.logical_id[1]))
791
        disk.physical_id = disk.logical_id = (disk.logical_id[0],
792
                                              os.path.join(file_storage_dir,
793
                                                           inst.name,
794
                                                           disk.iv_name))
795

    
796
    self._config_data.instances[inst.name] = inst
797
    self._WriteConfig()
798

    
799
  @locking.ssynchronized(_config_lock)
800
  def MarkInstanceDown(self, instance_name):
801
    """Mark the status of an instance to down in the configuration.
802

803
    """
804
    self._SetInstanceStatus(instance_name, False)
805

    
806
  def _UnlockedGetInstanceList(self):
807
    """Get the list of instances.
808

809
    This function is for internal use, when the config lock is already held.
810

811
    """
812
    return self._config_data.instances.keys()
813

    
814
  @locking.ssynchronized(_config_lock, shared=1)
815
  def GetInstanceList(self):
816
    """Get the list of instances.
817

818
    @return: array of instances, ex. ['instance2.example.com',
819
        'instance1.example.com']
820

821
    """
822
    return self._UnlockedGetInstanceList()
823

    
824
  @locking.ssynchronized(_config_lock, shared=1)
825
  def ExpandInstanceName(self, short_name):
826
    """Attempt to expand an incomplete instance name.
827

828
    """
829
    return utils.MatchNameComponent(short_name,
830
                                    self._config_data.instances.keys())
831

    
832
  def _UnlockedGetInstanceInfo(self, instance_name):
833
    """Returns information about an instance.
834

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

837
    """
838
    if instance_name not in self._config_data.instances:
839
      return None
840

    
841
    return self._config_data.instances[instance_name]
842

    
843
  @locking.ssynchronized(_config_lock, shared=1)
844
  def GetInstanceInfo(self, instance_name):
845
    """Returns information about an instance.
846

847
    It takes the information from the configuration file. Other information of
848
    an instance are taken from the live systems.
849

850
    @param instance_name: name of the instance, e.g.
851
        I{instance1.example.com}
852

853
    @rtype: L{objects.Instance}
854
    @return: the instance object
855

856
    """
857
    return self._UnlockedGetInstanceInfo(instance_name)
858

    
859
  @locking.ssynchronized(_config_lock, shared=1)
860
  def GetAllInstancesInfo(self):
861
    """Get the configuration of all instances.
862

863
    @rtype: dict
864
    @return: dict of (instance, instance_info), where instance_info is what
865
              would GetInstanceInfo return for the node
866

867
    """
868
    my_dict = dict([(instance, self._UnlockedGetInstanceInfo(instance))
869
                    for instance in self._UnlockedGetInstanceList()])
870
    return my_dict
871

    
872
  @locking.ssynchronized(_config_lock)
873
  def AddNode(self, node):
874
    """Add a node to the configuration.
875

876
    @type node: L{objects.Node}
877
    @param node: a Node instance
878

879
    """
880
    logging.info("Adding node %s to configuration" % node.name)
881

    
882
    node.serial_no = 1
883
    self._config_data.nodes[node.name] = node
884
    self._config_data.cluster.serial_no += 1
885
    self._WriteConfig()
886

    
887
  @locking.ssynchronized(_config_lock)
888
  def RemoveNode(self, node_name):
889
    """Remove a node from the configuration.
890

891
    """
892
    logging.info("Removing node %s from configuration" % node_name)
893

    
894
    if node_name not in self._config_data.nodes:
895
      raise errors.ConfigurationError("Unknown node '%s'" % node_name)
896

    
897
    del self._config_data.nodes[node_name]
898
    self._config_data.cluster.serial_no += 1
899
    self._WriteConfig()
900

    
901
  @locking.ssynchronized(_config_lock, shared=1)
902
  def ExpandNodeName(self, short_name):
903
    """Attempt to expand an incomplete instance name.
904

905
    """
906
    return utils.MatchNameComponent(short_name,
907
                                    self._config_data.nodes.keys())
908

    
909
  def _UnlockedGetNodeInfo(self, node_name):
910
    """Get the configuration of a node, as stored in the config.
911

912
    This function is for internal use, when the config lock is already
913
    held.
914

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

917
    @rtype: L{objects.Node}
918
    @return: the node object
919

920
    """
921
    if node_name not in self._config_data.nodes:
922
      return None
923

    
924
    return self._config_data.nodes[node_name]
925

    
926

    
927
  @locking.ssynchronized(_config_lock, shared=1)
928
  def GetNodeInfo(self, node_name):
929
    """Get the configuration of a node, as stored in the config.
930

931
    This is just a locked wrapper over L{_UnlockedGetNodeInfo}.
932

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

935
    @rtype: L{objects.Node}
936
    @return: the node object
937

938
    """
939
    return self._UnlockedGetNodeInfo(node_name)
940

    
941
  def _UnlockedGetNodeList(self):
942
    """Return the list of nodes which are in the configuration.
943

944
    This function is for internal use, when the config lock is already
945
    held.
946

947
    @rtype: list
948

949
    """
950
    return self._config_data.nodes.keys()
951

    
952

    
953
  @locking.ssynchronized(_config_lock, shared=1)
954
  def GetNodeList(self):
955
    """Return the list of nodes which are in the configuration.
956

957
    """
958
    return self._UnlockedGetNodeList()
959

    
960
  @locking.ssynchronized(_config_lock, shared=1)
961
  def GetOnlineNodeList(self):
962
    """Return the list of nodes which are online.
963

964
    """
965
    all_nodes = [self._UnlockedGetNodeInfo(node)
966
                 for node in self._UnlockedGetNodeList()]
967
    return [node.name for node in all_nodes if not node.offline]
968

    
969
  @locking.ssynchronized(_config_lock, shared=1)
970
  def GetAllNodesInfo(self):
971
    """Get the configuration of all nodes.
972

973
    @rtype: dict
974
    @return: dict of (node, node_info), where node_info is what
975
              would GetNodeInfo return for the node
976

977
    """
978
    my_dict = dict([(node, self._UnlockedGetNodeInfo(node))
979
                    for node in self._UnlockedGetNodeList()])
980
    return my_dict
981

    
982
  def _UnlockedGetMasterCandidateStats(self, exceptions=None):
983
    """Get the number of current and maximum desired and possible candidates.
984

985
    @type exceptions: list
986
    @param exceptions: if passed, list of nodes that should be ignored
987
    @rtype: tuple
988
    @return: tuple of (current, desired and possible)
989

990
    """
991
    mc_now = mc_max = 0
992
    for node in self._config_data.nodes.values():
993
      if exceptions and node.name in exceptions:
994
        continue
995
      if not (node.offline or node.drained):
996
        mc_max += 1
997
      if node.master_candidate:
998
        mc_now += 1
999
    mc_max = min(mc_max, self._config_data.cluster.candidate_pool_size)
1000
    return (mc_now, mc_max)
1001

    
1002
  @locking.ssynchronized(_config_lock, shared=1)
1003
  def GetMasterCandidateStats(self, exceptions=None):
1004
    """Get the number of current and maximum possible candidates.
1005

1006
    This is just a wrapper over L{_UnlockedGetMasterCandidateStats}.
1007

1008
    @type exceptions: list
1009
    @param exceptions: if passed, list of nodes that should be ignored
1010
    @rtype: tuple
1011
    @return: tuple of (current, max)
1012

1013
    """
1014
    return self._UnlockedGetMasterCandidateStats(exceptions)
1015

    
1016
  @locking.ssynchronized(_config_lock)
1017
  def MaintainCandidatePool(self):
1018
    """Try to grow the candidate pool to the desired size.
1019

1020
    @rtype: list
1021
    @return: list with the adjusted nodes (L{objects.Node} instances)
1022

1023
    """
1024
    mc_now, mc_max = self._UnlockedGetMasterCandidateStats()
1025
    mod_list = []
1026
    if mc_now < mc_max:
1027
      node_list = self._config_data.nodes.keys()
1028
      random.shuffle(node_list)
1029
      for name in node_list:
1030
        if mc_now >= mc_max:
1031
          break
1032
        node = self._config_data.nodes[name]
1033
        if node.master_candidate or node.offline or node.drained:
1034
          continue
1035
        mod_list.append(node)
1036
        node.master_candidate = True
1037
        node.serial_no += 1
1038
        mc_now += 1
1039
      if mc_now != mc_max:
1040
        # this should not happen
1041
        logging.warning("Warning: MaintainCandidatePool didn't manage to"
1042
                        " fill the candidate pool (%d/%d)", mc_now, mc_max)
1043
      if mod_list:
1044
        self._config_data.cluster.serial_no += 1
1045
        self._WriteConfig()
1046

    
1047
    return mod_list
1048

    
1049
  def _BumpSerialNo(self):
1050
    """Bump up the serial number of the config.
1051

1052
    """
1053
    self._config_data.serial_no += 1
1054

    
1055
  def _OpenConfig(self):
1056
    """Read the config data from disk.
1057

1058
    """
1059
    f = open(self._cfg_file, 'r')
1060
    try:
1061
      try:
1062
        data = objects.ConfigData.FromDict(serializer.Load(f.read()))
1063
      except Exception, err:
1064
        raise errors.ConfigurationError(err)
1065
    finally:
1066
      f.close()
1067

    
1068
    # Make sure the configuration has the right version
1069
    _ValidateConfig(data)
1070

    
1071
    if (not hasattr(data, 'cluster') or
1072
        not hasattr(data.cluster, 'rsahostkeypub')):
1073
      raise errors.ConfigurationError("Incomplete configuration"
1074
                                      " (missing cluster.rsahostkeypub)")
1075

    
1076
    # Upgrade configuration if needed
1077
    data.UpgradeConfig()
1078

    
1079
    self._config_data = data
1080
    # reset the last serial as -1 so that the next write will cause
1081
    # ssconf update
1082
    self._last_cluster_serial = -1
1083

    
1084
  def _DistributeConfig(self):
1085
    """Distribute the configuration to the other nodes.
1086

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

1090
    """
1091
    if self._offline:
1092
      return True
1093
    bad = False
1094

    
1095
    node_list = []
1096
    addr_list = []
1097
    myhostname = self._my_hostname
1098
    # we can skip checking whether _UnlockedGetNodeInfo returns None
1099
    # since the node list comes from _UnlocketGetNodeList, and we are
1100
    # called with the lock held, so no modifications should take place
1101
    # in between
1102
    for node_name in self._UnlockedGetNodeList():
1103
      if node_name == myhostname:
1104
        continue
1105
      node_info = self._UnlockedGetNodeInfo(node_name)
1106
      if not node_info.master_candidate:
1107
        continue
1108
      node_list.append(node_info.name)
1109
      addr_list.append(node_info.primary_ip)
1110

    
1111
    result = rpc.RpcRunner.call_upload_file(node_list, self._cfg_file,
1112
                                            address_list=addr_list)
1113
    for node in node_list:
1114
      if not result[node]:
1115
        logging.error("copy of file %s to node %s failed",
1116
                      self._cfg_file, node)
1117
        bad = True
1118
    return not bad
1119

    
1120
  def _WriteConfig(self, destination=None):
1121
    """Write the configuration data to persistent storage.
1122

1123
    """
1124
    # first, cleanup the _temporary_ids set, if an ID is now in the
1125
    # other objects it should be discarded to prevent unbounded growth
1126
    # of that structure
1127
    self._CleanupTemporaryIDs()
1128
    config_errors = self._UnlockedVerifyConfig()
1129
    if config_errors:
1130
      raise errors.ConfigurationError("Configuration data is not"
1131
                                      " consistent: %s" %
1132
                                      (", ".join(config_errors)))
1133
    if destination is None:
1134
      destination = self._cfg_file
1135
    self._BumpSerialNo()
1136
    txt = serializer.Dump(self._config_data.ToDict())
1137
    dir_name, file_name = os.path.split(destination)
1138
    fd, name = tempfile.mkstemp('.newconfig', file_name, dir_name)
1139
    f = os.fdopen(fd, 'w')
1140
    try:
1141
      f.write(txt)
1142
      os.fsync(f.fileno())
1143
    finally:
1144
      f.close()
1145
    # we don't need to do os.close(fd) as f.close() did it
1146
    os.rename(name, destination)
1147
    self.write_count += 1
1148

    
1149
    # and redistribute the config file to master candidates
1150
    self._DistributeConfig()
1151

    
1152
    # Write ssconf files on all nodes (including locally)
1153
    if self._last_cluster_serial < self._config_data.cluster.serial_no:
1154
      if not self._offline:
1155
        rpc.RpcRunner.call_write_ssconf_files(self._UnlockedGetNodeList(),
1156
                                              self._UnlockedGetSsconfValues())
1157
      self._last_cluster_serial = self._config_data.cluster.serial_no
1158

    
1159
  def _UnlockedGetSsconfValues(self):
1160
    """Return the values needed by ssconf.
1161

1162
    @rtype: dict
1163
    @return: a dictionary with keys the ssconf names and values their
1164
        associated value
1165

1166
    """
1167
    fn = "\n".join
1168
    instance_names = utils.NiceSort(self._UnlockedGetInstanceList())
1169
    node_names = utils.NiceSort(self._UnlockedGetNodeList())
1170
    node_info = [self._UnlockedGetNodeInfo(name) for name in node_names]
1171

    
1172
    instance_data = fn(instance_names)
1173
    off_data = fn(node.name for node in node_info if node.offline)
1174
    on_data = fn(node.name for node in node_info if not node.offline)
1175
    mc_data = fn(node.name for node in node_info if node.master_candidate)
1176
    node_data = fn(node_names)
1177

    
1178
    cluster = self._config_data.cluster
1179
    cluster_tags = fn(cluster.GetTags())
1180
    return {
1181
      constants.SS_CLUSTER_NAME: cluster.cluster_name,
1182
      constants.SS_CLUSTER_TAGS: cluster_tags,
1183
      constants.SS_FILE_STORAGE_DIR: cluster.file_storage_dir,
1184
      constants.SS_MASTER_CANDIDATES: mc_data,
1185
      constants.SS_MASTER_IP: cluster.master_ip,
1186
      constants.SS_MASTER_NETDEV: cluster.master_netdev,
1187
      constants.SS_MASTER_NODE: cluster.master_node,
1188
      constants.SS_NODE_LIST: node_data,
1189
      constants.SS_OFFLINE_NODES: off_data,
1190
      constants.SS_ONLINE_NODES: on_data,
1191
      constants.SS_INSTANCE_LIST: instance_data,
1192
      constants.SS_RELEASE_VERSION: constants.RELEASE_VERSION,
1193
      }
1194

    
1195
  @locking.ssynchronized(_config_lock, shared=1)
1196
  def GetVGName(self):
1197
    """Return the volume group name.
1198

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

    
1202
  @locking.ssynchronized(_config_lock)
1203
  def SetVGName(self, vg_name):
1204
    """Set the volume group name.
1205

1206
    """
1207
    self._config_data.cluster.volume_group_name = vg_name
1208
    self._config_data.cluster.serial_no += 1
1209
    self._WriteConfig()
1210

    
1211
  @locking.ssynchronized(_config_lock, shared=1)
1212
  def GetDefBridge(self):
1213
    """Return the default bridge.
1214

1215
    """
1216
    return self._config_data.cluster.default_bridge
1217

    
1218
  @locking.ssynchronized(_config_lock, shared=1)
1219
  def GetMACPrefix(self):
1220
    """Return the mac prefix.
1221

1222
    """
1223
    return self._config_data.cluster.mac_prefix
1224

    
1225
  @locking.ssynchronized(_config_lock, shared=1)
1226
  def GetClusterInfo(self):
1227
    """Returns information about the cluster
1228

1229
    @rtype: L{objects.Cluster}
1230
    @return: the cluster object
1231

1232
    """
1233
    return self._config_data.cluster
1234

    
1235
  @locking.ssynchronized(_config_lock)
1236
  def Update(self, target):
1237
    """Notify function to be called after updates.
1238

1239
    This function must be called when an object (as returned by
1240
    GetInstanceInfo, GetNodeInfo, GetCluster) has been updated and the
1241
    caller wants the modifications saved to the backing store. Note
1242
    that all modified objects will be saved, but the target argument
1243
    is the one the caller wants to ensure that it's saved.
1244

1245
    @param target: an instance of either L{objects.Cluster},
1246
        L{objects.Node} or L{objects.Instance} which is existing in
1247
        the cluster
1248

1249
    """
1250
    if self._config_data is None:
1251
      raise errors.ProgrammerError("Configuration file not read,"
1252
                                   " cannot save.")
1253
    update_serial = False
1254
    if isinstance(target, objects.Cluster):
1255
      test = target == self._config_data.cluster
1256
    elif isinstance(target, objects.Node):
1257
      test = target in self._config_data.nodes.values()
1258
      update_serial = True
1259
    elif isinstance(target, objects.Instance):
1260
      test = target in self._config_data.instances.values()
1261
    else:
1262
      raise errors.ProgrammerError("Invalid object type (%s) passed to"
1263
                                   " ConfigWriter.Update" % type(target))
1264
    if not test:
1265
      raise errors.ConfigurationError("Configuration updated since object"
1266
                                      " has been read or unknown object")
1267
    target.serial_no += 1
1268

    
1269
    if update_serial:
1270
      # for node updates, we need to increase the cluster serial too
1271
      self._config_data.cluster.serial_no += 1
1272

    
1273
    if isinstance(target, objects.Instance):
1274
      self._UnlockedReleaseDRBDMinors(target.name)
1275
      for nic in target.nics:
1276
        self._temporary_macs.discard(nic.mac)
1277

    
1278
    self._WriteConfig()