Statistics
| Branch: | Tag: | Revision:

root / lib / config.py @ 34e54ebc

History | View | Annotate | Download (41.1 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
    return existing
179

    
180
  @locking.ssynchronized(_config_lock, shared=1)
181
  def GenerateUniqueID(self, exceptions=None):
182
    """Generate an unique disk name.
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 checked
188
        for uniqueness (used for example when you want to get
189
        more than one id at one time without adding each one in
190
        turn 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
  def _AllMACs(self):
211
    """Return all MACs present in the config.
212

213
    @rtype: list
214
    @return: the list of all MACs
215

216
    """
217
    result = []
218
    for instance in self._config_data.instances.values():
219
      for nic in instance.nics:
220
        result.append(nic.mac)
221

    
222
    return result
223

    
224
  def _AllDRBDSecrets(self):
225
    """Return all DRBD secrets present in the config.
226

227
    @rtype: list
228
    @return: the list of all DRBD secrets
229

230
    """
231
    def helper(disk, result):
232
      """Recursively gather secrets from this disk."""
233
      if disk.dev_type == constants.DT_DRBD8:
234
        result.append(disk.logical_id[5])
235
      if disk.children:
236
        for child in disk.children:
237
          helper(child, result)
238

    
239
    result = []
240
    for instance in self._config_data.instances.values():
241
      for disk in instance.disks:
242
        helper(disk, result)
243

    
244
    return result
245

    
246
  def _CheckDiskIDs(self, disk, l_ids, p_ids):
247
    """Compute duplicate disk IDs
248

249
    @type disk: L{objects.Disk}
250
    @param disk: the disk at which to start searching
251
    @type l_ids: list
252
    @param l_ids: list of current logical ids
253
    @type p_ids: list
254
    @param p_ids: list of current physical ids
255
    @rtype: list
256
    @return: a list of error messages
257

258
    """
259
    result = []
260
    if disk.logical_id is not None:
261
      if disk.logical_id in l_ids:
262
        result.append("duplicate logical id %s" % str(disk.logical_id))
263
      else:
264
        l_ids.append(disk.logical_id)
265
    if disk.physical_id is not None:
266
      if disk.physical_id in p_ids:
267
        result.append("duplicate physical id %s" % str(disk.physical_id))
268
      else:
269
        p_ids.append(disk.physical_id)
270

    
271
    if disk.children:
272
      for child in disk.children:
273
        result.extend(self._CheckDiskIDs(child, l_ids, p_ids))
274
    return result
275

    
276
  def _UnlockedVerifyConfig(self):
277
    """Verify function.
278

279
    @rtype: list
280
    @return: a list of error messages; a non-empty list signifies
281
        configuration errors
282

283
    """
284
    result = []
285
    seen_macs = []
286
    ports = {}
287
    data = self._config_data
288
    seen_lids = []
289
    seen_pids = []
290

    
291
    # global cluster checks
292
    if not data.cluster.enabled_hypervisors:
293
      result.append("enabled hypervisors list doesn't have any entries")
294
    invalid_hvs = set(data.cluster.enabled_hypervisors) - constants.HYPER_TYPES
295
    if invalid_hvs:
296
      result.append("enabled hypervisors contains invalid entries: %s" %
297
                    invalid_hvs)
298

    
299
    if data.cluster.master_node not in data.nodes:
300
      result.append("cluster has invalid primary node '%s'" %
301
                    data.cluster.master_node)
302

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

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

    
334
      # instance disk verify
335
      for idx, disk in enumerate(instance.disks):
336
        result.extend(["instance '%s' disk %d error: %s" %
337
                       (instance.name, idx, msg) for msg in disk.Verify()])
338
        result.extend(self._CheckDiskIDs(disk, seen_lids, seen_pids))
339

    
340
    # cluster-wide pool of free ports
341
    for free_port in data.cluster.tcpudp_port_pool:
342
      if free_port not in ports:
343
        ports[free_port] = []
344
      ports[free_port].append(("cluster", "port marked as free"))
345

    
346
    # compute tcp/udp duplicate ports
347
    keys = ports.keys()
348
    keys.sort()
349
    for pnum in keys:
350
      pdata = ports[pnum]
351
      if len(pdata) > 1:
352
        txt = ", ".join(["%s/%s" % val for val in pdata])
353
        result.append("tcp/udp port %s has duplicates: %s" % (pnum, txt))
354

    
355
    # highest used tcp port check
356
    if keys:
357
      if keys[-1] > data.cluster.highest_used_port:
358
        result.append("Highest used port mismatch, saved %s, computed %s" %
359
                      (data.cluster.highest_used_port, keys[-1]))
360

    
361
    if not data.nodes[data.cluster.master_node].master_candidate:
362
      result.append("Master node is not a master candidate")
363

    
364
    # master candidate checks
365
    mc_now, mc_max = self._UnlockedGetMasterCandidateStats()
366
    if mc_now < mc_max:
367
      result.append("Not enough master candidates: actual %d, target %d" %
368
                    (mc_now, mc_max))
369

    
370
    # node checks
371
    for node in data.nodes.values():
372
      if [node.master_candidate, node.drained, node.offline].count(True) > 1:
373
        result.append("Node %s state is invalid: master_candidate=%s,"
374
                      " drain=%s, offline=%s" %
375
                      (node.name, node.master_candidate, node.drain,
376
                       node.offline))
377

    
378
    # drbd minors check
379
    d_map, duplicates = self._UnlockedComputeDRBDMap()
380
    for node, minor, instance_a, instance_b in duplicates:
381
      result.append("DRBD minor %d on node %s is assigned twice to instances"
382
                    " %s and %s" % (minor, node, instance_a, instance_b))
383

    
384
    return result
385

    
386
  @locking.ssynchronized(_config_lock, shared=1)
387
  def VerifyConfig(self):
388
    """Verify function.
389

390
    This is just a wrapper over L{_UnlockedVerifyConfig}.
391

392
    @rtype: list
393
    @return: a list of error messages; a non-empty list signifies
394
        configuration errors
395

396
    """
397
    return self._UnlockedVerifyConfig()
398

    
399
  def _UnlockedSetDiskID(self, disk, node_name):
400
    """Convert the unique ID to the ID needed on the target nodes.
401

402
    This is used only for drbd, which needs ip/port configuration.
403

404
    The routine descends down and updates its children also, because
405
    this helps when the only the top device is passed to the remote
406
    node.
407

408
    This function is for internal use, when the config lock is already held.
409

410
    """
411
    if disk.children:
412
      for child in disk.children:
413
        self._UnlockedSetDiskID(child, node_name)
414

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

    
437
  @locking.ssynchronized(_config_lock)
438
  def SetDiskID(self, disk, node_name):
439
    """Convert the unique ID to the ID needed on the target nodes.
440

441
    This is used only for drbd, which needs ip/port configuration.
442

443
    The routine descends down and updates its children also, because
444
    this helps when the only the top device is passed to the remote
445
    node.
446

447
    """
448
    return self._UnlockedSetDiskID(disk, node_name)
449

    
450
  @locking.ssynchronized(_config_lock)
451
  def AddTcpUdpPort(self, port):
452
    """Adds a new port to the available port pool.
453

454
    """
455
    if not isinstance(port, int):
456
      raise errors.ProgrammerError("Invalid type passed for port")
457

    
458
    self._config_data.cluster.tcpudp_port_pool.add(port)
459
    self._WriteConfig()
460

    
461
  @locking.ssynchronized(_config_lock, shared=1)
462
  def GetPortList(self):
463
    """Returns a copy of the current port list.
464

465
    """
466
    return self._config_data.cluster.tcpudp_port_pool.copy()
467

    
468
  @locking.ssynchronized(_config_lock)
469
  def AllocatePort(self):
470
    """Allocate a port.
471

472
    The port will be taken from the available port pool or from the
473
    default port range (and in this case we increase
474
    highest_used_port).
475

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

    
488
    self._WriteConfig()
489
    return port
490

    
491
  def _UnlockedComputeDRBDMap(self):
492
    """Compute the used DRBD minor/nodes.
493

494
    @rtype: (dict, list)
495
    @return: dictionary of node_name: dict of minor: instance_name;
496
        the returned dict will have all the nodes in it (even if with
497
        an empty list), and a list of duplicates; if the duplicates
498
        list is not empty, the configuration is corrupted and its caller
499
        should raise an exception
500

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

    
518
    duplicates = []
519
    my_dict = dict((node, {}) for node in self._config_data.nodes)
520
    for instance in self._config_data.instances.itervalues():
521
      for disk in instance.disks:
522
        duplicates.extend(_AppendUsedPorts(instance.name, disk, my_dict))
523
    for (node, minor), instance in self._temporary_drbds.iteritems():
524
      if minor in my_dict[node] and my_dict[node][minor] != instance:
525
        duplicates.append((node, minor, instance, my_dict[node][minor]))
526
      else:
527
        my_dict[node][minor] = instance
528
    return my_dict, duplicates
529

    
530
  @locking.ssynchronized(_config_lock)
531
  def ComputeDRBDMap(self):
532
    """Compute the used DRBD minor/nodes.
533

534
    This is just a wrapper over L{_UnlockedComputeDRBDMap}.
535

536
    @return: dictionary of node_name: dict of minor: instance_name;
537
        the returned dict will have all the nodes in it (even if with
538
        an empty list).
539

540
    """
541
    d_map, duplicates = self._UnlockedComputeDRBDMap()
542
    if duplicates:
543
      raise errors.ConfigurationError("Duplicate DRBD ports detected: %s" %
544
                                      str(duplicates))
545
    return d_map
546

    
547
  @locking.ssynchronized(_config_lock)
548
  def AllocateDRBDMinor(self, nodes, instance):
549
    """Allocate a drbd minor.
550

551
    The free minor will be automatically computed from the existing
552
    devices. A node can be given multiple times in order to allocate
553
    multiple minors. The result is the list of minors, in the same
554
    order as the passed nodes.
555

556
    @type instance: string
557
    @param instance: the instance for which we allocate minors
558

559
    """
560
    assert isinstance(instance, basestring), \
561
           "Invalid argument '%s' passed to AllocateDRBDMinor" % instance
562

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

    
603
  def _UnlockedReleaseDRBDMinors(self, instance):
604
    """Release temporary drbd minors allocated for a given instance.
605

606
    @type instance: string
607
    @param instance: the instance for which temporary minors should be
608
                     released
609

610
    """
611
    assert isinstance(instance, basestring), \
612
           "Invalid argument passed to ReleaseDRBDMinors"
613
    for key, name in self._temporary_drbds.items():
614
      if name == instance:
615
        del self._temporary_drbds[key]
616

    
617
  @locking.ssynchronized(_config_lock)
618
  def ReleaseDRBDMinors(self, instance):
619
    """Release temporary drbd minors allocated for a given instance.
620

621
    This should be called on the error paths, on the success paths
622
    it's automatically called by the ConfigWriter add and update
623
    functions.
624

625
    This function is just a wrapper over L{_UnlockedReleaseDRBDMinors}.
626

627
    @type instance: string
628
    @param instance: the instance for which temporary minors should be
629
                     released
630

631
    """
632
    self._UnlockedReleaseDRBDMinors(instance)
633

    
634
  @locking.ssynchronized(_config_lock, shared=1)
635
  def GetConfigVersion(self):
636
    """Get the configuration version.
637

638
    @return: Config version
639

640
    """
641
    return self._config_data.version
642

    
643
  @locking.ssynchronized(_config_lock, shared=1)
644
  def GetClusterName(self):
645
    """Get cluster name.
646

647
    @return: Cluster name
648

649
    """
650
    return self._config_data.cluster.cluster_name
651

    
652
  @locking.ssynchronized(_config_lock, shared=1)
653
  def GetMasterNode(self):
654
    """Get the hostname of the master node for this cluster.
655

656
    @return: Master hostname
657

658
    """
659
    return self._config_data.cluster.master_node
660

    
661
  @locking.ssynchronized(_config_lock, shared=1)
662
  def GetMasterIP(self):
663
    """Get the IP of the master node for this cluster.
664

665
    @return: Master IP
666

667
    """
668
    return self._config_data.cluster.master_ip
669

    
670
  @locking.ssynchronized(_config_lock, shared=1)
671
  def GetMasterNetdev(self):
672
    """Get the master network device for this cluster.
673

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

    
677
  @locking.ssynchronized(_config_lock, shared=1)
678
  def GetFileStorageDir(self):
679
    """Get the file storage dir for this cluster.
680

681
    """
682
    return self._config_data.cluster.file_storage_dir
683

    
684
  @locking.ssynchronized(_config_lock, shared=1)
685
  def GetHypervisorType(self):
686
    """Get the hypervisor type for this cluster.
687

688
    """
689
    return self._config_data.cluster.enabled_hypervisors[0]
690

    
691
  @locking.ssynchronized(_config_lock, shared=1)
692
  def GetHostKey(self):
693
    """Return the rsa hostkey from the config.
694

695
    @rtype: string
696
    @return: the rsa hostkey
697

698
    """
699
    return self._config_data.cluster.rsahostkeypub
700

    
701
  @locking.ssynchronized(_config_lock)
702
  def AddInstance(self, instance):
703
    """Add an instance to the config.
704

705
    This should be used after creating a new instance.
706

707
    @type instance: L{objects.Instance}
708
    @param instance: the instance object
709

710
    """
711
    if not isinstance(instance, objects.Instance):
712
      raise errors.ProgrammerError("Invalid type passed to AddInstance")
713

    
714
    if instance.disk_template != constants.DT_DISKLESS:
715
      all_lvs = instance.MapLVsByNode()
716
      logging.info("Instance '%s' DISK_LAYOUT: %s", instance.name, all_lvs)
717

    
718
    all_macs = self._AllMACs()
719
    for nic in instance.nics:
720
      if nic.mac in all_macs:
721
        raise errors.ConfigurationError("Cannot add instance %s:"
722
          " MAC address '%s' already in use." % (instance.name, nic.mac))
723

    
724
    instance.serial_no = 1
725
    instance.ctime = instance.mtime = time.time()
726
    self._config_data.instances[instance.name] = instance
727
    self._config_data.cluster.serial_no += 1
728
    self._UnlockedReleaseDRBDMinors(instance.name)
729
    for nic in instance.nics:
730
      self._temporary_macs.discard(nic.mac)
731
    self._WriteConfig()
732

    
733
  def _SetInstanceStatus(self, instance_name, status):
734
    """Set the instance's status to a given value.
735

736
    """
737
    assert isinstance(status, bool), \
738
           "Invalid status '%s' passed to SetInstanceStatus" % (status,)
739

    
740
    if instance_name not in self._config_data.instances:
741
      raise errors.ConfigurationError("Unknown instance '%s'" %
742
                                      instance_name)
743
    instance = self._config_data.instances[instance_name]
744
    if instance.admin_up != status:
745
      instance.admin_up = status
746
      instance.serial_no += 1
747
      instance.mtime = time.time()
748
      self._WriteConfig()
749

    
750
  @locking.ssynchronized(_config_lock)
751
  def MarkInstanceUp(self, instance_name):
752
    """Mark the instance status to up in the config.
753

754
    """
755
    self._SetInstanceStatus(instance_name, True)
756

    
757
  @locking.ssynchronized(_config_lock)
758
  def RemoveInstance(self, instance_name):
759
    """Remove the instance from the configuration.
760

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

    
768
  @locking.ssynchronized(_config_lock)
769
  def RenameInstance(self, old_name, new_name):
770
    """Rename an instance.
771

772
    This needs to be done in ConfigWriter and not by RemoveInstance
773
    combined with AddInstance as only we can guarantee an atomic
774
    rename.
775

776
    """
777
    if old_name not in self._config_data.instances:
778
      raise errors.ConfigurationError("Unknown instance '%s'" % old_name)
779
    inst = self._config_data.instances[old_name]
780
    del self._config_data.instances[old_name]
781
    inst.name = new_name
782

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

    
792
    self._config_data.instances[inst.name] = inst
793
    self._WriteConfig()
794

    
795
  @locking.ssynchronized(_config_lock)
796
  def MarkInstanceDown(self, instance_name):
797
    """Mark the status of an instance to down in the configuration.
798

799
    """
800
    self._SetInstanceStatus(instance_name, False)
801

    
802
  def _UnlockedGetInstanceList(self):
803
    """Get the list of instances.
804

805
    This function is for internal use, when the config lock is already held.
806

807
    """
808
    return self._config_data.instances.keys()
809

    
810
  @locking.ssynchronized(_config_lock, shared=1)
811
  def GetInstanceList(self):
812
    """Get the list of instances.
813

814
    @return: array of instances, ex. ['instance2.example.com',
815
        'instance1.example.com']
816

817
    """
818
    return self._UnlockedGetInstanceList()
819

    
820
  @locking.ssynchronized(_config_lock, shared=1)
821
  def ExpandInstanceName(self, short_name):
822
    """Attempt to expand an incomplete instance name.
823

824
    """
825
    return utils.MatchNameComponent(short_name,
826
                                    self._config_data.instances.keys())
827

    
828
  def _UnlockedGetInstanceInfo(self, instance_name):
829
    """Returns information about an instance.
830

831
    This function is for internal use, when the config lock is already held.
832

833
    """
834
    if instance_name not in self._config_data.instances:
835
      return None
836

    
837
    return self._config_data.instances[instance_name]
838

    
839
  @locking.ssynchronized(_config_lock, shared=1)
840
  def GetInstanceInfo(self, instance_name):
841
    """Returns information about an instance.
842

843
    It takes the information from the configuration file. Other information of
844
    an instance are taken from the live systems.
845

846
    @param instance_name: name of the instance, e.g.
847
        I{instance1.example.com}
848

849
    @rtype: L{objects.Instance}
850
    @return: the instance object
851

852
    """
853
    return self._UnlockedGetInstanceInfo(instance_name)
854

    
855
  @locking.ssynchronized(_config_lock, shared=1)
856
  def GetAllInstancesInfo(self):
857
    """Get the configuration of all instances.
858

859
    @rtype: dict
860
    @return: dict of (instance, instance_info), where instance_info is what
861
              would GetInstanceInfo return for the node
862

863
    """
864
    my_dict = dict([(instance, self._UnlockedGetInstanceInfo(instance))
865
                    for instance in self._UnlockedGetInstanceList()])
866
    return my_dict
867

    
868
  @locking.ssynchronized(_config_lock)
869
  def AddNode(self, node):
870
    """Add a node to the configuration.
871

872
    @type node: L{objects.Node}
873
    @param node: a Node instance
874

875
    """
876
    logging.info("Adding node %s to configuration" % node.name)
877

    
878
    node.serial_no = 1
879
    node.ctime = node.mtime = time.time()
880
    self._config_data.nodes[node.name] = node
881
    self._config_data.cluster.serial_no += 1
882
    self._WriteConfig()
883

    
884
  @locking.ssynchronized(_config_lock)
885
  def RemoveNode(self, node_name):
886
    """Remove a node from the configuration.
887

888
    """
889
    logging.info("Removing node %s from configuration" % node_name)
890

    
891
    if node_name not in self._config_data.nodes:
892
      raise errors.ConfigurationError("Unknown node '%s'" % node_name)
893

    
894
    del self._config_data.nodes[node_name]
895
    self._config_data.cluster.serial_no += 1
896
    self._WriteConfig()
897

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

902
    """
903
    return utils.MatchNameComponent(short_name,
904
                                    self._config_data.nodes.keys())
905

    
906
  def _UnlockedGetNodeInfo(self, node_name):
907
    """Get the configuration of a node, as stored in the config.
908

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

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

914
    @rtype: L{objects.Node}
915
    @return: the node object
916

917
    """
918
    if node_name not in self._config_data.nodes:
919
      return None
920

    
921
    return self._config_data.nodes[node_name]
922

    
923

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

928
    This is just a locked wrapper over L{_UnlockedGetNodeInfo}.
929

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

932
    @rtype: L{objects.Node}
933
    @return: the node object
934

935
    """
936
    return self._UnlockedGetNodeInfo(node_name)
937

    
938
  def _UnlockedGetNodeList(self):
939
    """Return the list of nodes which are in the configuration.
940

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

944
    @rtype: list
945

946
    """
947
    return self._config_data.nodes.keys()
948

    
949

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

954
    """
955
    return self._UnlockedGetNodeList()
956

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

961
    """
962
    all_nodes = [self._UnlockedGetNodeInfo(node)
963
                 for node in self._UnlockedGetNodeList()]
964
    return [node.name for node in all_nodes if not node.offline]
965

    
966
  @locking.ssynchronized(_config_lock, shared=1)
967
  def GetAllNodesInfo(self):
968
    """Get the configuration of all nodes.
969

970
    @rtype: dict
971
    @return: dict of (node, node_info), where node_info is what
972
              would GetNodeInfo return for the node
973

974
    """
975
    my_dict = dict([(node, self._UnlockedGetNodeInfo(node))
976
                    for node in self._UnlockedGetNodeList()])
977
    return my_dict
978

    
979
  def _UnlockedGetMasterCandidateStats(self, exceptions=None):
980
    """Get the number of current and maximum desired and possible candidates.
981

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

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

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

1003
    This is just a wrapper over L{_UnlockedGetMasterCandidateStats}.
1004

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

1010
    """
1011
    return self._UnlockedGetMasterCandidateStats(exceptions)
1012

    
1013
  @locking.ssynchronized(_config_lock)
1014
  def MaintainCandidatePool(self):
1015
    """Try to grow the candidate pool to the desired size.
1016

1017
    @rtype: list
1018
    @return: list with the adjusted nodes (L{objects.Node} instances)
1019

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

    
1044
    return mod_list
1045

    
1046
  def _BumpSerialNo(self):
1047
    """Bump up the serial number of the config.
1048

1049
    """
1050
    self._config_data.serial_no += 1
1051
    self._config_data.mtime = time.time()
1052

    
1053
  def _OpenConfig(self):
1054
    """Read the config data from disk.
1055

1056
    """
1057
    raw_data = utils.ReadFile(self._cfg_file)
1058

    
1059
    try:
1060
      data = objects.ConfigData.FromDict(serializer.Load(raw_data))
1061
    except Exception, err:
1062
      raise errors.ConfigurationError(err)
1063

    
1064
    # Make sure the configuration has the right version
1065
    _ValidateConfig(data)
1066

    
1067
    if (not hasattr(data, 'cluster') or
1068
        not hasattr(data.cluster, 'rsahostkeypub')):
1069
      raise errors.ConfigurationError("Incomplete configuration"
1070
                                      " (missing cluster.rsahostkeypub)")
1071

    
1072
    # Upgrade configuration if needed
1073
    data.UpgradeConfig()
1074

    
1075
    self._config_data = data
1076
    # reset the last serial as -1 so that the next write will cause
1077
    # ssconf update
1078
    self._last_cluster_serial = -1
1079

    
1080
  def _DistributeConfig(self):
1081
    """Distribute the configuration to the other nodes.
1082

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

1086
    """
1087
    if self._offline:
1088
      return True
1089
    bad = False
1090

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

    
1107
    result = rpc.RpcRunner.call_upload_file(node_list, self._cfg_file,
1108
                                            address_list=addr_list)
1109
    for to_node, to_result in result.items():
1110
      msg = to_result.fail_msg
1111
      if msg:
1112
        msg = ("Copy of file %s to node %s failed: %s" %
1113
               (self._cfg_file, to_node, msg))
1114
        logging.error(msg)
1115
        bad = True
1116
    return not bad
1117

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

1121
    """
1122
    config_errors = self._UnlockedVerifyConfig()
1123
    if config_errors:
1124
      raise errors.ConfigurationError("Configuration data is not"
1125
                                      " consistent: %s" %
1126
                                      (", ".join(config_errors)))
1127
    if destination is None:
1128
      destination = self._cfg_file
1129
    self._BumpSerialNo()
1130
    txt = serializer.Dump(self._config_data.ToDict())
1131

    
1132
    utils.WriteFile(destination, data=txt)
1133

    
1134
    self.write_count += 1
1135

    
1136
    # and redistribute the config file to master candidates
1137
    self._DistributeConfig()
1138

    
1139
    # Write ssconf files on all nodes (including locally)
1140
    if self._last_cluster_serial < self._config_data.cluster.serial_no:
1141
      if not self._offline:
1142
        result = rpc.RpcRunner.call_write_ssconf_files(\
1143
          self._UnlockedGetNodeList(),
1144
          self._UnlockedGetSsconfValues())
1145
        for nname, nresu in result.items():
1146
          msg = nresu.fail_msg
1147
          if msg:
1148
            logging.warning("Error while uploading ssconf files to"
1149
                            " node %s: %s", nname, msg)
1150
      self._last_cluster_serial = self._config_data.cluster.serial_no
1151

    
1152
  def _UnlockedGetSsconfValues(self):
1153
    """Return the values needed by ssconf.
1154

1155
    @rtype: dict
1156
    @return: a dictionary with keys the ssconf names and values their
1157
        associated value
1158

1159
    """
1160
    fn = "\n".join
1161
    instance_names = utils.NiceSort(self._UnlockedGetInstanceList())
1162
    node_names = utils.NiceSort(self._UnlockedGetNodeList())
1163
    node_info = [self._UnlockedGetNodeInfo(name) for name in node_names]
1164
    node_pri_ips = ["%s %s" % (ninfo.name, ninfo.primary_ip)
1165
                    for ninfo in node_info]
1166
    node_snd_ips = ["%s %s" % (ninfo.name, ninfo.secondary_ip)
1167
                    for ninfo in node_info]
1168

    
1169
    instance_data = fn(instance_names)
1170
    off_data = fn(node.name for node in node_info if node.offline)
1171
    on_data = fn(node.name for node in node_info if not node.offline)
1172
    mc_data = fn(node.name for node in node_info if node.master_candidate)
1173
    mc_ips_data = fn(node.primary_ip for node in node_info
1174
                     if node.master_candidate)
1175
    node_data = fn(node_names)
1176
    node_pri_ips_data = fn(node_pri_ips)
1177
    node_snd_ips_data = fn(node_snd_ips)
1178

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

    
1199
  @locking.ssynchronized(_config_lock, shared=1)
1200
  def GetVGName(self):
1201
    """Return the volume group name.
1202

1203
    """
1204
    return self._config_data.cluster.volume_group_name
1205

    
1206
  @locking.ssynchronized(_config_lock)
1207
  def SetVGName(self, vg_name):
1208
    """Set the volume group name.
1209

1210
    """
1211
    self._config_data.cluster.volume_group_name = vg_name
1212
    self._config_data.cluster.serial_no += 1
1213
    self._WriteConfig()
1214

    
1215
  @locking.ssynchronized(_config_lock, shared=1)
1216
  def GetMACPrefix(self):
1217
    """Return the mac prefix.
1218

1219
    """
1220
    return self._config_data.cluster.mac_prefix
1221

    
1222
  @locking.ssynchronized(_config_lock, shared=1)
1223
  def GetClusterInfo(self):
1224
    """Returns information about the cluster
1225

1226
    @rtype: L{objects.Cluster}
1227
    @return: the cluster object
1228

1229
    """
1230
    return self._config_data.cluster
1231

    
1232
  @locking.ssynchronized(_config_lock)
1233
  def Update(self, target):
1234
    """Notify function to be called after updates.
1235

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

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

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

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

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

    
1277
    self._WriteConfig()