Statistics
| Branch: | Tag: | Revision:

root / lib / config.py @ afa1386e

History | View | Annotate | Download (45.7 kB)

1
#
2
#
3

    
4
# Copyright (C) 2006, 2007 Google Inc.
5
#
6
# This program is free software; you can redistribute it and/or modify
7
# it under the terms of the GNU General Public License as published by
8
# the Free Software Foundation; either version 2 of the License, or
9
# (at your option) any later version.
10
#
11
# This program is distributed in the hope that it will be useful, but
12
# WITHOUT ANY WARRANTY; without even the implied warranty of
13
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
14
# General Public License for more details.
15
#
16
# You should have received a copy of the GNU General Public License
17
# along with this program; if not, write to the Free Software
18
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
19
# 02110-1301, USA.
20

    
21

    
22
"""Configuration management for Ganeti
23

24
This module provides the interface to the Ganeti cluster configuration.
25

26
The configuration data is stored on every node but is updated on the master
27
only. After each update, the master distributes the data to the other nodes.
28

29
Currently, the data storage format is JSON. YAML was slow and consuming too
30
much memory.
31

32
"""
33

    
34
import os
35
import random
36
import logging
37
import time
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
# job id used for resource management at config upgrade time
51
_UPGRADE_CONFIG_JID="jid-cfg-upgrade"
52

    
53

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

57
  This only verifies the version of the configuration.
58

59
  @raise errors.ConfigurationError: if the version differs from what
60
      we expect
61

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

    
69

    
70
class TemporaryReservationManager:
71
  """A temporary resource reservation manager.
72

73
  This is used to reserve resources in a job, before using them, making sure
74
  other jobs cannot get them in the meantime.
75

76
  """
77
  def __init__(self):
78
    self._ec_reserved = {}
79

    
80
  def Reserved(self, resource):
81
    for holder_reserved in self._ec_reserved.items():
82
      if resource in holder_reserved:
83
        return True
84
    return False
85

    
86
  def Reserve(self, ec_id, resource):
87
    if self.Reserved(resource):
88
      raise errors.ReservationError("Duplicate reservation for resource: %s." %
89
                                    (resource))
90
    if ec_id not in self._ec_reserved:
91
      self._ec_reserved[ec_id] = set([resource])
92
    else:
93
      self._ec_reserved[ec_id].add(resource)
94

    
95
  def DropECReservations(self, ec_id):
96
    if ec_id in self._ec_reserved:
97
      del self._ec_reserved[ec_id]
98

    
99
  def GetReserved(self):
100
    all_reserved = set()
101
    for holder_reserved in self._ec_reserved.values():
102
      all_reserved.update(holder_reserved)
103
    return all_reserved
104

    
105
  def Generate(self, existing, generate_one_fn, ec_id):
106
    """Generate a new resource of this type
107

108
    """
109
    assert callable(generate_one_fn)
110

    
111
    all_elems = self.GetReserved()
112
    all_elems.update(existing)
113
    retries = 64
114
    while retries > 0:
115
      new_resource = generate_one_fn()
116
      if new_resource is not None and new_resource not in all_elems:
117
        break
118
    else:
119
      raise errors.ConfigurationError("Not able generate new resource"
120
                                      " (last tried: %s)" % new_resource)
121
    self.Reserve(ec_id, new_resource)
122
    return new_resource
123

    
124

    
125
class ConfigWriter:
126
  """The interface to the cluster configuration.
127

128
  """
129
  def __init__(self, cfg_file=None, offline=False):
130
    self.write_count = 0
131
    self._lock = _config_lock
132
    self._config_data = None
133
    self._offline = offline
134
    if cfg_file is None:
135
      self._cfg_file = constants.CLUSTER_CONF_FILE
136
    else:
137
      self._cfg_file = cfg_file
138
    self._temporary_ids = TemporaryReservationManager()
139
    self._temporary_drbds = {}
140
    self._temporary_macs = TemporaryReservationManager()
141
    self._temporary_secrets = TemporaryReservationManager()
142
    # Note: in order to prevent errors when resolving our name in
143
    # _DistributeConfig, we compute it here once and reuse it; it's
144
    # better to raise an error before starting to modify the config
145
    # file than after it was modified
146
    self._my_hostname = utils.HostInfo().name
147
    self._last_cluster_serial = -1
148
    self._OpenConfig()
149

    
150
  # this method needs to be static, so that we can call it on the class
151
  @staticmethod
152
  def IsCluster():
153
    """Check if the cluster is configured.
154

155
    """
156
    return os.path.exists(constants.CLUSTER_CONF_FILE)
157

    
158
  def _GenerateOneMAC(self):
159
    """Generate one mac address
160

161
    """
162
    prefix = self._config_data.cluster.mac_prefix
163
    byte1 = random.randrange(0, 256)
164
    byte2 = random.randrange(0, 256)
165
    byte3 = random.randrange(0, 256)
166
    mac = "%s:%02x:%02x:%02x" % (prefix, byte1, byte2, byte3)
167
    return mac
168

    
169
  @locking.ssynchronized(_config_lock, shared=1)
170
  def GenerateMAC(self, ec_id):
171
    """Generate a MAC for an instance.
172

173
    This should check the current instances for duplicates.
174

175
    """
176
    existing = self._AllMACs()
177
    return self._temporary_ids.Generate(existing, self._GenerateOneMAC, ec_id)
178

    
179
  @locking.ssynchronized(_config_lock, shared=1)
180
  def ReserveMAC(self, mac, ec_id):
181
    """Reserve a MAC for an instance.
182

183
    This only checks instances managed by this cluster, it does not
184
    check for potential collisions elsewhere.
185

186
    """
187
    all_macs = self._AllMACs()
188
    if mac in all_macs:
189
      raise errors.ReservationError("mac already in use")
190
    else:
191
      self._temporary_macs.Reserve(mac, ec_id)
192

    
193
  @locking.ssynchronized(_config_lock, shared=1)
194
  def GenerateDRBDSecret(self, ec_id):
195
    """Generate a DRBD secret.
196

197
    This checks the current disks for duplicates.
198

199
    """
200
    return self._temporary_secrets.Generate(self._AllDRBDSecrets(),
201
                                            utils.GenerateSecret,
202
                                            ec_id)
203
  def _AllLVs(self):
204
    """Compute the list of all LVs.
205

206
    """
207
    lvnames = set()
208
    for instance in self._config_data.instances.values():
209
      node_data = instance.MapLVsByNode()
210
      for lv_list in node_data.values():
211
        lvnames.update(lv_list)
212
    return lvnames
213

    
214
  def _AllIDs(self, include_temporary):
215
    """Compute the list of all UUIDs and names we have.
216

217
    @type include_temporary: boolean
218
    @param include_temporary: whether to include the _temporary_ids set
219
    @rtype: set
220
    @return: a set of IDs
221

222
    """
223
    existing = set()
224
    if include_temporary:
225
      existing.update(self._temporary_ids.GetReserved())
226
    existing.update(self._AllLVs())
227
    existing.update(self._config_data.instances.keys())
228
    existing.update(self._config_data.nodes.keys())
229
    existing.update([i.uuid for i in self._AllUUIDObjects() if i.uuid])
230
    return existing
231

    
232
  def _GenerateUniqueID(self, ec_id):
233
    """Generate an unique UUID.
234

235
    This checks the current node, instances and disk names for
236
    duplicates.
237

238
    @rtype: string
239
    @return: the unique id
240

241
    """
242
    existing = self._AllIDs(include_temporary=False)
243
    return self._temporary_ids.Generate(existing, utils.NewUUID, ec_id)
244

    
245
  @locking.ssynchronized(_config_lock, shared=1)
246
  def GenerateUniqueID(self, ec_id):
247
    """Generate an unique ID.
248

249
    This is just a wrapper over the unlocked version.
250

251
    @type ec_id: string
252
    @param ec_id: unique id for the job to reserve the id to
253

254
    """
255
    return self._GenerateUniqueID(ec_id)
256

    
257
  def _AllMACs(self):
258
    """Return all MACs present in the config.
259

260
    @rtype: list
261
    @return: the list of all MACs
262

263
    """
264
    result = []
265
    for instance in self._config_data.instances.values():
266
      for nic in instance.nics:
267
        result.append(nic.mac)
268

    
269
    return result
270

    
271
  def _AllDRBDSecrets(self):
272
    """Return all DRBD secrets present in the config.
273

274
    @rtype: list
275
    @return: the list of all DRBD secrets
276

277
    """
278
    def helper(disk, result):
279
      """Recursively gather secrets from this disk."""
280
      if disk.dev_type == constants.DT_DRBD8:
281
        result.append(disk.logical_id[5])
282
      if disk.children:
283
        for child in disk.children:
284
          helper(child, result)
285

    
286
    result = []
287
    for instance in self._config_data.instances.values():
288
      for disk in instance.disks:
289
        helper(disk, result)
290

    
291
    return result
292

    
293
  def _CheckDiskIDs(self, disk, l_ids, p_ids):
294
    """Compute duplicate disk IDs
295

296
    @type disk: L{objects.Disk}
297
    @param disk: the disk at which to start searching
298
    @type l_ids: list
299
    @param l_ids: list of current logical ids
300
    @type p_ids: list
301
    @param p_ids: list of current physical ids
302
    @rtype: list
303
    @return: a list of error messages
304

305
    """
306
    result = []
307
    if disk.logical_id is not None:
308
      if disk.logical_id in l_ids:
309
        result.append("duplicate logical id %s" % str(disk.logical_id))
310
      else:
311
        l_ids.append(disk.logical_id)
312
    if disk.physical_id is not None:
313
      if disk.physical_id in p_ids:
314
        result.append("duplicate physical id %s" % str(disk.physical_id))
315
      else:
316
        p_ids.append(disk.physical_id)
317

    
318
    if disk.children:
319
      for child in disk.children:
320
        result.extend(self._CheckDiskIDs(child, l_ids, p_ids))
321
    return result
322

    
323
  def _UnlockedVerifyConfig(self):
324
    """Verify function.
325

326
    @rtype: list
327
    @return: a list of error messages; a non-empty list signifies
328
        configuration errors
329

330
    """
331
    result = []
332
    seen_macs = []
333
    ports = {}
334
    data = self._config_data
335
    seen_lids = []
336
    seen_pids = []
337

    
338
    # global cluster checks
339
    if not data.cluster.enabled_hypervisors:
340
      result.append("enabled hypervisors list doesn't have any entries")
341
    invalid_hvs = set(data.cluster.enabled_hypervisors) - constants.HYPER_TYPES
342
    if invalid_hvs:
343
      result.append("enabled hypervisors contains invalid entries: %s" %
344
                    invalid_hvs)
345

    
346
    if data.cluster.master_node not in data.nodes:
347
      result.append("cluster has invalid primary node '%s'" %
348
                    data.cluster.master_node)
349

    
350
    # per-instance checks
351
    for instance_name in data.instances:
352
      instance = data.instances[instance_name]
353
      if instance.primary_node not in data.nodes:
354
        result.append("instance '%s' has invalid primary node '%s'" %
355
                      (instance_name, instance.primary_node))
356
      for snode in instance.secondary_nodes:
357
        if snode not in data.nodes:
358
          result.append("instance '%s' has invalid secondary node '%s'" %
359
                        (instance_name, snode))
360
      for idx, nic in enumerate(instance.nics):
361
        if nic.mac in seen_macs:
362
          result.append("instance '%s' has NIC %d mac %s duplicate" %
363
                        (instance_name, idx, nic.mac))
364
        else:
365
          seen_macs.append(nic.mac)
366

    
367
      # gather the drbd ports for duplicate checks
368
      for dsk in instance.disks:
369
        if dsk.dev_type in constants.LDS_DRBD:
370
          tcp_port = dsk.logical_id[2]
371
          if tcp_port not in ports:
372
            ports[tcp_port] = []
373
          ports[tcp_port].append((instance.name, "drbd disk %s" % dsk.iv_name))
374
      # gather network port reservation
375
      net_port = getattr(instance, "network_port", None)
376
      if net_port is not None:
377
        if net_port not in ports:
378
          ports[net_port] = []
379
        ports[net_port].append((instance.name, "network port"))
380

    
381
      # instance disk verify
382
      for idx, disk in enumerate(instance.disks):
383
        result.extend(["instance '%s' disk %d error: %s" %
384
                       (instance.name, idx, msg) for msg in disk.Verify()])
385
        result.extend(self._CheckDiskIDs(disk, seen_lids, seen_pids))
386

    
387
    # cluster-wide pool of free ports
388
    for free_port in data.cluster.tcpudp_port_pool:
389
      if free_port not in ports:
390
        ports[free_port] = []
391
      ports[free_port].append(("cluster", "port marked as free"))
392

    
393
    # compute tcp/udp duplicate ports
394
    keys = ports.keys()
395
    keys.sort()
396
    for pnum in keys:
397
      pdata = ports[pnum]
398
      if len(pdata) > 1:
399
        txt = ", ".join(["%s/%s" % val for val in pdata])
400
        result.append("tcp/udp port %s has duplicates: %s" % (pnum, txt))
401

    
402
    # highest used tcp port check
403
    if keys:
404
      if keys[-1] > data.cluster.highest_used_port:
405
        result.append("Highest used port mismatch, saved %s, computed %s" %
406
                      (data.cluster.highest_used_port, keys[-1]))
407

    
408
    if not data.nodes[data.cluster.master_node].master_candidate:
409
      result.append("Master node is not a master candidate")
410

    
411
    # master candidate checks
412
    mc_now, mc_max, _ = self._UnlockedGetMasterCandidateStats()
413
    if mc_now < mc_max:
414
      result.append("Not enough master candidates: actual %d, target %d" %
415
                    (mc_now, mc_max))
416

    
417
    # node checks
418
    for node in data.nodes.values():
419
      if [node.master_candidate, node.drained, node.offline].count(True) > 1:
420
        result.append("Node %s state is invalid: master_candidate=%s,"
421
                      " drain=%s, offline=%s" %
422
                      (node.name, node.master_candidate, node.drain,
423
                       node.offline))
424

    
425
    # drbd minors check
426
    d_map, duplicates = self._UnlockedComputeDRBDMap()
427
    for node, minor, instance_a, instance_b in duplicates:
428
      result.append("DRBD minor %d on node %s is assigned twice to instances"
429
                    " %s and %s" % (minor, node, instance_a, instance_b))
430

    
431
    # IP checks
432
    ips = { data.cluster.master_ip: ["cluster_ip"] }
433
    def _helper(ip, name):
434
      if ip in ips:
435
        ips[ip].append(name)
436
      else:
437
        ips[ip] = [name]
438

    
439
    for node in data.nodes.values():
440
      _helper(node.primary_ip, "node:%s/primary" % node.name)
441
      if node.secondary_ip != node.primary_ip:
442
        _helper(node.secondary_ip, "node:%s/secondary" % node.name)
443

    
444
    for ip, owners in ips.items():
445
      if len(owners) > 1:
446
        result.append("IP address %s is used by multiple owners: %s" %
447
                      (ip, ", ".join(owners)))
448
    return result
449

    
450
  @locking.ssynchronized(_config_lock, shared=1)
451
  def VerifyConfig(self):
452
    """Verify function.
453

454
    This is just a wrapper over L{_UnlockedVerifyConfig}.
455

456
    @rtype: list
457
    @return: a list of error messages; a non-empty list signifies
458
        configuration errors
459

460
    """
461
    return self._UnlockedVerifyConfig()
462

    
463
  def _UnlockedSetDiskID(self, disk, node_name):
464
    """Convert the unique ID to the ID needed on the target nodes.
465

466
    This is used only for drbd, which needs ip/port configuration.
467

468
    The routine descends down and updates its children also, because
469
    this helps when the only the top device is passed to the remote
470
    node.
471

472
    This function is for internal use, when the config lock is already held.
473

474
    """
475
    if disk.children:
476
      for child in disk.children:
477
        self._UnlockedSetDiskID(child, node_name)
478

    
479
    if disk.logical_id is None and disk.physical_id is not None:
480
      return
481
    if disk.dev_type == constants.LD_DRBD8:
482
      pnode, snode, port, pminor, sminor, secret = disk.logical_id
483
      if node_name not in (pnode, snode):
484
        raise errors.ConfigurationError("DRBD device not knowing node %s" %
485
                                        node_name)
486
      pnode_info = self._UnlockedGetNodeInfo(pnode)
487
      snode_info = self._UnlockedGetNodeInfo(snode)
488
      if pnode_info is None or snode_info is None:
489
        raise errors.ConfigurationError("Can't find primary or secondary node"
490
                                        " for %s" % str(disk))
491
      p_data = (pnode_info.secondary_ip, port)
492
      s_data = (snode_info.secondary_ip, port)
493
      if pnode == node_name:
494
        disk.physical_id = p_data + s_data + (pminor, secret)
495
      else: # it must be secondary, we tested above
496
        disk.physical_id = s_data + p_data + (sminor, secret)
497
    else:
498
      disk.physical_id = disk.logical_id
499
    return
500

    
501
  @locking.ssynchronized(_config_lock)
502
  def SetDiskID(self, disk, node_name):
503
    """Convert the unique ID to the ID needed on the target nodes.
504

505
    This is used only for drbd, which needs ip/port configuration.
506

507
    The routine descends down and updates its children also, because
508
    this helps when the only the top device is passed to the remote
509
    node.
510

511
    """
512
    return self._UnlockedSetDiskID(disk, node_name)
513

    
514
  @locking.ssynchronized(_config_lock)
515
  def AddTcpUdpPort(self, port):
516
    """Adds a new port to the available port pool.
517

518
    """
519
    if not isinstance(port, int):
520
      raise errors.ProgrammerError("Invalid type passed for port")
521

    
522
    self._config_data.cluster.tcpudp_port_pool.add(port)
523
    self._WriteConfig()
524

    
525
  @locking.ssynchronized(_config_lock, shared=1)
526
  def GetPortList(self):
527
    """Returns a copy of the current port list.
528

529
    """
530
    return self._config_data.cluster.tcpudp_port_pool.copy()
531

    
532
  @locking.ssynchronized(_config_lock)
533
  def AllocatePort(self):
534
    """Allocate a port.
535

536
    The port will be taken from the available port pool or from the
537
    default port range (and in this case we increase
538
    highest_used_port).
539

540
    """
541
    # If there are TCP/IP ports configured, we use them first.
542
    if self._config_data.cluster.tcpudp_port_pool:
543
      port = self._config_data.cluster.tcpudp_port_pool.pop()
544
    else:
545
      port = self._config_data.cluster.highest_used_port + 1
546
      if port >= constants.LAST_DRBD_PORT:
547
        raise errors.ConfigurationError("The highest used port is greater"
548
                                        " than %s. Aborting." %
549
                                        constants.LAST_DRBD_PORT)
550
      self._config_data.cluster.highest_used_port = port
551

    
552
    self._WriteConfig()
553
    return port
554

    
555
  def _UnlockedComputeDRBDMap(self):
556
    """Compute the used DRBD minor/nodes.
557

558
    @rtype: (dict, list)
559
    @return: dictionary of node_name: dict of minor: instance_name;
560
        the returned dict will have all the nodes in it (even if with
561
        an empty list), and a list of duplicates; if the duplicates
562
        list is not empty, the configuration is corrupted and its caller
563
        should raise an exception
564

565
    """
566
    def _AppendUsedPorts(instance_name, disk, used):
567
      duplicates = []
568
      if disk.dev_type == constants.LD_DRBD8 and len(disk.logical_id) >= 5:
569
        node_a, node_b, _, minor_a, minor_b = disk.logical_id[:5]
570
        for node, port in ((node_a, minor_a), (node_b, minor_b)):
571
          assert node in used, ("Node '%s' of instance '%s' not found"
572
                                " in node list" % (node, instance_name))
573
          if port in used[node]:
574
            duplicates.append((node, port, instance_name, used[node][port]))
575
          else:
576
            used[node][port] = instance_name
577
      if disk.children:
578
        for child in disk.children:
579
          duplicates.extend(_AppendUsedPorts(instance_name, child, used))
580
      return duplicates
581

    
582
    duplicates = []
583
    my_dict = dict((node, {}) for node in self._config_data.nodes)
584
    for instance in self._config_data.instances.itervalues():
585
      for disk in instance.disks:
586
        duplicates.extend(_AppendUsedPorts(instance.name, disk, my_dict))
587
    for (node, minor), instance in self._temporary_drbds.iteritems():
588
      if minor in my_dict[node] and my_dict[node][minor] != instance:
589
        duplicates.append((node, minor, instance, my_dict[node][minor]))
590
      else:
591
        my_dict[node][minor] = instance
592
    return my_dict, duplicates
593

    
594
  @locking.ssynchronized(_config_lock)
595
  def ComputeDRBDMap(self):
596
    """Compute the used DRBD minor/nodes.
597

598
    This is just a wrapper over L{_UnlockedComputeDRBDMap}.
599

600
    @return: dictionary of node_name: dict of minor: instance_name;
601
        the returned dict will have all the nodes in it (even if with
602
        an empty list).
603

604
    """
605
    d_map, duplicates = self._UnlockedComputeDRBDMap()
606
    if duplicates:
607
      raise errors.ConfigurationError("Duplicate DRBD ports detected: %s" %
608
                                      str(duplicates))
609
    return d_map
610

    
611
  @locking.ssynchronized(_config_lock)
612
  def AllocateDRBDMinor(self, nodes, instance):
613
    """Allocate a drbd minor.
614

615
    The free minor will be automatically computed from the existing
616
    devices. A node can be given multiple times in order to allocate
617
    multiple minors. The result is the list of minors, in the same
618
    order as the passed nodes.
619

620
    @type instance: string
621
    @param instance: the instance for which we allocate minors
622

623
    """
624
    assert isinstance(instance, basestring), \
625
           "Invalid argument '%s' passed to AllocateDRBDMinor" % instance
626

    
627
    d_map, duplicates = self._UnlockedComputeDRBDMap()
628
    if duplicates:
629
      raise errors.ConfigurationError("Duplicate DRBD ports detected: %s" %
630
                                      str(duplicates))
631
    result = []
632
    for nname in nodes:
633
      ndata = d_map[nname]
634
      if not ndata:
635
        # no minors used, we can start at 0
636
        result.append(0)
637
        ndata[0] = instance
638
        self._temporary_drbds[(nname, 0)] = instance
639
        continue
640
      keys = ndata.keys()
641
      keys.sort()
642
      ffree = utils.FirstFree(keys)
643
      if ffree is None:
644
        # return the next minor
645
        # TODO: implement high-limit check
646
        minor = keys[-1] + 1
647
      else:
648
        minor = ffree
649
      # double-check minor against current instances
650
      assert minor not in d_map[nname], \
651
             ("Attempt to reuse allocated DRBD minor %d on node %s,"
652
              " already allocated to instance %s" %
653
              (minor, nname, d_map[nname][minor]))
654
      ndata[minor] = instance
655
      # double-check minor against reservation
656
      r_key = (nname, minor)
657
      assert r_key not in self._temporary_drbds, \
658
             ("Attempt to reuse reserved DRBD minor %d on node %s,"
659
              " reserved for instance %s" %
660
              (minor, nname, self._temporary_drbds[r_key]))
661
      self._temporary_drbds[r_key] = instance
662
      result.append(minor)
663
    logging.debug("Request to allocate drbd minors, input: %s, returning %s",
664
                  nodes, result)
665
    return result
666

    
667
  def _UnlockedReleaseDRBDMinors(self, instance):
668
    """Release temporary drbd minors allocated for a given instance.
669

670
    @type instance: string
671
    @param instance: the instance for which temporary minors should be
672
                     released
673

674
    """
675
    assert isinstance(instance, basestring), \
676
           "Invalid argument passed to ReleaseDRBDMinors"
677
    for key, name in self._temporary_drbds.items():
678
      if name == instance:
679
        del self._temporary_drbds[key]
680

    
681
  @locking.ssynchronized(_config_lock)
682
  def ReleaseDRBDMinors(self, instance):
683
    """Release temporary drbd minors allocated for a given instance.
684

685
    This should be called on the error paths, on the success paths
686
    it's automatically called by the ConfigWriter add and update
687
    functions.
688

689
    This function is just a wrapper over L{_UnlockedReleaseDRBDMinors}.
690

691
    @type instance: string
692
    @param instance: the instance for which temporary minors should be
693
                     released
694

695
    """
696
    self._UnlockedReleaseDRBDMinors(instance)
697

    
698
  @locking.ssynchronized(_config_lock, shared=1)
699
  def GetConfigVersion(self):
700
    """Get the configuration version.
701

702
    @return: Config version
703

704
    """
705
    return self._config_data.version
706

    
707
  @locking.ssynchronized(_config_lock, shared=1)
708
  def GetClusterName(self):
709
    """Get cluster name.
710

711
    @return: Cluster name
712

713
    """
714
    return self._config_data.cluster.cluster_name
715

    
716
  @locking.ssynchronized(_config_lock, shared=1)
717
  def GetMasterNode(self):
718
    """Get the hostname of the master node for this cluster.
719

720
    @return: Master hostname
721

722
    """
723
    return self._config_data.cluster.master_node
724

    
725
  @locking.ssynchronized(_config_lock, shared=1)
726
  def GetMasterIP(self):
727
    """Get the IP of the master node for this cluster.
728

729
    @return: Master IP
730

731
    """
732
    return self._config_data.cluster.master_ip
733

    
734
  @locking.ssynchronized(_config_lock, shared=1)
735
  def GetMasterNetdev(self):
736
    """Get the master network device for this cluster.
737

738
    """
739
    return self._config_data.cluster.master_netdev
740

    
741
  @locking.ssynchronized(_config_lock, shared=1)
742
  def GetFileStorageDir(self):
743
    """Get the file storage dir for this cluster.
744

745
    """
746
    return self._config_data.cluster.file_storage_dir
747

    
748
  @locking.ssynchronized(_config_lock, shared=1)
749
  def GetHypervisorType(self):
750
    """Get the hypervisor type for this cluster.
751

752
    """
753
    return self._config_data.cluster.enabled_hypervisors[0]
754

    
755
  @locking.ssynchronized(_config_lock, shared=1)
756
  def GetHostKey(self):
757
    """Return the rsa hostkey from the config.
758

759
    @rtype: string
760
    @return: the rsa hostkey
761

762
    """
763
    return self._config_data.cluster.rsahostkeypub
764

    
765
  @locking.ssynchronized(_config_lock)
766
  def AddInstance(self, instance, ec_id):
767
    """Add an instance to the config.
768

769
    This should be used after creating a new instance.
770

771
    @type instance: L{objects.Instance}
772
    @param instance: the instance object
773

774
    """
775
    if not isinstance(instance, objects.Instance):
776
      raise errors.ProgrammerError("Invalid type passed to AddInstance")
777

    
778
    if instance.disk_template != constants.DT_DISKLESS:
779
      all_lvs = instance.MapLVsByNode()
780
      logging.info("Instance '%s' DISK_LAYOUT: %s", instance.name, all_lvs)
781

    
782
    all_macs = self._AllMACs()
783
    for nic in instance.nics:
784
      if nic.mac in all_macs:
785
        raise errors.ConfigurationError("Cannot add instance %s:"
786
                                        " MAC address '%s' already in use." %
787
                                        (instance.name, nic.mac))
788

    
789
    self._EnsureUUID(instance, ec_id)
790

    
791
    instance.serial_no = 1
792
    instance.ctime = instance.mtime = time.time()
793
    self._config_data.instances[instance.name] = instance
794
    self._config_data.cluster.serial_no += 1
795
    self._UnlockedReleaseDRBDMinors(instance.name)
796
    self._WriteConfig()
797

    
798
  def _EnsureUUID(self, item, ec_id):
799
    """Ensures a given object has a valid UUID.
800

801
    @param item: the instance or node to be checked
802
    @param ec_id: the execution context id for the uuid reservation
803

804
    """
805
    if not item.uuid:
806
      item.uuid = self._GenerateUniqueID(ec_id)
807
    elif item.uuid in self._AllIDs(temporary=True):
808
      raise errors.ConfigurationError("Cannot add '%s': UUID already in use" %
809
                                      (item.name, item.uuid))
810

    
811
  def _SetInstanceStatus(self, instance_name, status):
812
    """Set the instance's status to a given value.
813

814
    """
815
    assert isinstance(status, bool), \
816
           "Invalid status '%s' passed to SetInstanceStatus" % (status,)
817

    
818
    if instance_name not in self._config_data.instances:
819
      raise errors.ConfigurationError("Unknown instance '%s'" %
820
                                      instance_name)
821
    instance = self._config_data.instances[instance_name]
822
    if instance.admin_up != status:
823
      instance.admin_up = status
824
      instance.serial_no += 1
825
      instance.mtime = time.time()
826
      self._WriteConfig()
827

    
828
  @locking.ssynchronized(_config_lock)
829
  def MarkInstanceUp(self, instance_name):
830
    """Mark the instance status to up in the config.
831

832
    """
833
    self._SetInstanceStatus(instance_name, True)
834

    
835
  @locking.ssynchronized(_config_lock)
836
  def RemoveInstance(self, instance_name):
837
    """Remove the instance from the configuration.
838

839
    """
840
    if instance_name not in self._config_data.instances:
841
      raise errors.ConfigurationError("Unknown instance '%s'" % instance_name)
842
    del self._config_data.instances[instance_name]
843
    self._config_data.cluster.serial_no += 1
844
    self._WriteConfig()
845

    
846
  @locking.ssynchronized(_config_lock)
847
  def RenameInstance(self, old_name, new_name):
848
    """Rename an instance.
849

850
    This needs to be done in ConfigWriter and not by RemoveInstance
851
    combined with AddInstance as only we can guarantee an atomic
852
    rename.
853

854
    """
855
    if old_name not in self._config_data.instances:
856
      raise errors.ConfigurationError("Unknown instance '%s'" % old_name)
857
    inst = self._config_data.instances[old_name]
858
    del self._config_data.instances[old_name]
859
    inst.name = new_name
860

    
861
    for disk in inst.disks:
862
      if disk.dev_type == constants.LD_FILE:
863
        # rename the file paths in logical and physical id
864
        file_storage_dir = os.path.dirname(os.path.dirname(disk.logical_id[1]))
865
        disk.physical_id = disk.logical_id = (disk.logical_id[0],
866
                                              os.path.join(file_storage_dir,
867
                                                           inst.name,
868
                                                           disk.iv_name))
869

    
870
    self._config_data.instances[inst.name] = inst
871
    self._WriteConfig()
872

    
873
  @locking.ssynchronized(_config_lock)
874
  def MarkInstanceDown(self, instance_name):
875
    """Mark the status of an instance to down in the configuration.
876

877
    """
878
    self._SetInstanceStatus(instance_name, False)
879

    
880
  def _UnlockedGetInstanceList(self):
881
    """Get the list of instances.
882

883
    This function is for internal use, when the config lock is already held.
884

885
    """
886
    return self._config_data.instances.keys()
887

    
888
  @locking.ssynchronized(_config_lock, shared=1)
889
  def GetInstanceList(self):
890
    """Get the list of instances.
891

892
    @return: array of instances, ex. ['instance2.example.com',
893
        'instance1.example.com']
894

895
    """
896
    return self._UnlockedGetInstanceList()
897

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

902
    """
903
    return utils.MatchNameComponent(short_name,
904
                                    self._config_data.instances.keys(),
905
                                    case_sensitive=False)
906

    
907
  def _UnlockedGetInstanceInfo(self, instance_name):
908
    """Returns information about an instance.
909

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

912
    """
913
    if instance_name not in self._config_data.instances:
914
      return None
915

    
916
    return self._config_data.instances[instance_name]
917

    
918
  @locking.ssynchronized(_config_lock, shared=1)
919
  def GetInstanceInfo(self, instance_name):
920
    """Returns information about an instance.
921

922
    It takes the information from the configuration file. Other information of
923
    an instance are taken from the live systems.
924

925
    @param instance_name: name of the instance, e.g.
926
        I{instance1.example.com}
927

928
    @rtype: L{objects.Instance}
929
    @return: the instance object
930

931
    """
932
    return self._UnlockedGetInstanceInfo(instance_name)
933

    
934
  @locking.ssynchronized(_config_lock, shared=1)
935
  def GetAllInstancesInfo(self):
936
    """Get the configuration of all instances.
937

938
    @rtype: dict
939
    @return: dict of (instance, instance_info), where instance_info is what
940
              would GetInstanceInfo return for the node
941

942
    """
943
    my_dict = dict([(instance, self._UnlockedGetInstanceInfo(instance))
944
                    for instance in self._UnlockedGetInstanceList()])
945
    return my_dict
946

    
947
  @locking.ssynchronized(_config_lock)
948
  def AddNode(self, node, ec_id):
949
    """Add a node to the configuration.
950

951
    @type node: L{objects.Node}
952
    @param node: a Node instance
953

954
    """
955
    logging.info("Adding node %s to configuration", node.name)
956

    
957
    self._EnsureUUID(node, ec_id)
958

    
959
    node.serial_no = 1
960
    node.ctime = node.mtime = time.time()
961
    self._config_data.nodes[node.name] = node
962
    self._config_data.cluster.serial_no += 1
963
    self._WriteConfig()
964

    
965
  @locking.ssynchronized(_config_lock)
966
  def RemoveNode(self, node_name):
967
    """Remove a node from the configuration.
968

969
    """
970
    logging.info("Removing node %s from configuration", node_name)
971

    
972
    if node_name not in self._config_data.nodes:
973
      raise errors.ConfigurationError("Unknown node '%s'" % node_name)
974

    
975
    del self._config_data.nodes[node_name]
976
    self._config_data.cluster.serial_no += 1
977
    self._WriteConfig()
978

    
979
  @locking.ssynchronized(_config_lock, shared=1)
980
  def ExpandNodeName(self, short_name):
981
    """Attempt to expand an incomplete instance name.
982

983
    """
984
    return utils.MatchNameComponent(short_name,
985
                                    self._config_data.nodes.keys(),
986
                                    case_sensitive=False)
987

    
988
  def _UnlockedGetNodeInfo(self, node_name):
989
    """Get the configuration of a node, as stored in the config.
990

991
    This function is for internal use, when the config lock is already
992
    held.
993

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

996
    @rtype: L{objects.Node}
997
    @return: the node object
998

999
    """
1000
    if node_name not in self._config_data.nodes:
1001
      return None
1002

    
1003
    return self._config_data.nodes[node_name]
1004

    
1005

    
1006
  @locking.ssynchronized(_config_lock, shared=1)
1007
  def GetNodeInfo(self, node_name):
1008
    """Get the configuration of a node, as stored in the config.
1009

1010
    This is just a locked wrapper over L{_UnlockedGetNodeInfo}.
1011

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

1014
    @rtype: L{objects.Node}
1015
    @return: the node object
1016

1017
    """
1018
    return self._UnlockedGetNodeInfo(node_name)
1019

    
1020
  def _UnlockedGetNodeList(self):
1021
    """Return the list of nodes which are in the configuration.
1022

1023
    This function is for internal use, when the config lock is already
1024
    held.
1025

1026
    @rtype: list
1027

1028
    """
1029
    return self._config_data.nodes.keys()
1030

    
1031

    
1032
  @locking.ssynchronized(_config_lock, shared=1)
1033
  def GetNodeList(self):
1034
    """Return the list of nodes which are in the configuration.
1035

1036
    """
1037
    return self._UnlockedGetNodeList()
1038

    
1039
  @locking.ssynchronized(_config_lock, shared=1)
1040
  def GetOnlineNodeList(self):
1041
    """Return the list of nodes which are online.
1042

1043
    """
1044
    all_nodes = [self._UnlockedGetNodeInfo(node)
1045
                 for node in self._UnlockedGetNodeList()]
1046
    return [node.name for node in all_nodes if not node.offline]
1047

    
1048
  @locking.ssynchronized(_config_lock, shared=1)
1049
  def GetAllNodesInfo(self):
1050
    """Get the configuration of all nodes.
1051

1052
    @rtype: dict
1053
    @return: dict of (node, node_info), where node_info is what
1054
              would GetNodeInfo return for the node
1055

1056
    """
1057
    my_dict = dict([(node, self._UnlockedGetNodeInfo(node))
1058
                    for node in self._UnlockedGetNodeList()])
1059
    return my_dict
1060

    
1061
  def _UnlockedGetMasterCandidateStats(self, exceptions=None):
1062
    """Get the number of current and maximum desired and possible candidates.
1063

1064
    @type exceptions: list
1065
    @param exceptions: if passed, list of nodes that should be ignored
1066
    @rtype: tuple
1067
    @return: tuple of (current, desired and possible, possible)
1068

1069
    """
1070
    mc_now = mc_should = mc_max = 0
1071
    for node in self._config_data.nodes.values():
1072
      if exceptions and node.name in exceptions:
1073
        continue
1074
      if not (node.offline or node.drained):
1075
        mc_max += 1
1076
      if node.master_candidate:
1077
        mc_now += 1
1078
    mc_should = min(mc_max, self._config_data.cluster.candidate_pool_size)
1079
    return (mc_now, mc_should, mc_max)
1080

    
1081
  @locking.ssynchronized(_config_lock, shared=1)
1082
  def GetMasterCandidateStats(self, exceptions=None):
1083
    """Get the number of current and maximum possible candidates.
1084

1085
    This is just a wrapper over L{_UnlockedGetMasterCandidateStats}.
1086

1087
    @type exceptions: list
1088
    @param exceptions: if passed, list of nodes that should be ignored
1089
    @rtype: tuple
1090
    @return: tuple of (current, max)
1091

1092
    """
1093
    return self._UnlockedGetMasterCandidateStats(exceptions)
1094

    
1095
  @locking.ssynchronized(_config_lock)
1096
  def MaintainCandidatePool(self, exceptions):
1097
    """Try to grow the candidate pool to the desired size.
1098

1099
    @type exceptions: list
1100
    @param exceptions: if passed, list of nodes that should be ignored
1101
    @rtype: list
1102
    @return: list with the adjusted nodes (L{objects.Node} instances)
1103

1104
    """
1105
    mc_now, mc_max, _ = self._UnlockedGetMasterCandidateStats(exceptions)
1106
    mod_list = []
1107
    if mc_now < mc_max:
1108
      node_list = self._config_data.nodes.keys()
1109
      random.shuffle(node_list)
1110
      for name in node_list:
1111
        if mc_now >= mc_max:
1112
          break
1113
        node = self._config_data.nodes[name]
1114
        if (node.master_candidate or node.offline or node.drained or
1115
            node.name in exceptions):
1116
          continue
1117
        mod_list.append(node)
1118
        node.master_candidate = True
1119
        node.serial_no += 1
1120
        mc_now += 1
1121
      if mc_now != mc_max:
1122
        # this should not happen
1123
        logging.warning("Warning: MaintainCandidatePool didn't manage to"
1124
                        " fill the candidate pool (%d/%d)", mc_now, mc_max)
1125
      if mod_list:
1126
        self._config_data.cluster.serial_no += 1
1127
        self._WriteConfig()
1128

    
1129
    return mod_list
1130

    
1131
  def _BumpSerialNo(self):
1132
    """Bump up the serial number of the config.
1133

1134
    """
1135
    self._config_data.serial_no += 1
1136
    self._config_data.mtime = time.time()
1137

    
1138
  def _AllUUIDObjects(self):
1139
    """Returns all objects with uuid attributes.
1140

1141
    """
1142
    return (self._config_data.instances.values() +
1143
            self._config_data.nodes.values() +
1144
            [self._config_data.cluster])
1145

    
1146
  def _OpenConfig(self):
1147
    """Read the config data from disk.
1148

1149
    """
1150
    raw_data = utils.ReadFile(self._cfg_file)
1151

    
1152
    try:
1153
      data = objects.ConfigData.FromDict(serializer.Load(raw_data))
1154
    except Exception, err:
1155
      raise errors.ConfigurationError(err)
1156

    
1157
    # Make sure the configuration has the right version
1158
    _ValidateConfig(data)
1159

    
1160
    if (not hasattr(data, 'cluster') or
1161
        not hasattr(data.cluster, 'rsahostkeypub')):
1162
      raise errors.ConfigurationError("Incomplete configuration"
1163
                                      " (missing cluster.rsahostkeypub)")
1164

    
1165
    # Upgrade configuration if needed
1166
    data.UpgradeConfig()
1167

    
1168
    self._config_data = data
1169
    # reset the last serial as -1 so that the next write will cause
1170
    # ssconf update
1171
    self._last_cluster_serial = -1
1172

    
1173
    # And finally run our (custom) config upgrade sequence
1174
    self._UpgradeConfig()
1175

    
1176
  def _UpgradeConfig(self):
1177
    """Run upgrade steps that cannot be done purely in the objects.
1178

1179
    This is because some data elements need uniqueness across the
1180
    whole configuration, etc.
1181

1182
    @warning: this function will call L{_WriteConfig()}, so it needs
1183
        to either be called with the lock held or from a safe place
1184
        (the constructor)
1185

1186
    """
1187
    modified = False
1188
    for item in self._AllUUIDObjects():
1189
      if item.uuid is None:
1190
        item.uuid = self._GenerateUniqueID(_UPGRADE_CONFIG_JID)
1191
        modified = True
1192
    if modified:
1193
      self._WriteConfig()
1194
      # This is ok even if it acquires the internal lock, as _UpgradeConfig is
1195
      # only called at config init time, without the lock held
1196
      self.DropECReservations(_UPGRADE_CONFIG_JID)
1197

    
1198

    
1199
  def _DistributeConfig(self, feedback_fn):
1200
    """Distribute the configuration to the other nodes.
1201

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

1205
    """
1206
    if self._offline:
1207
      return True
1208

    
1209
    bad = False
1210

    
1211
    node_list = []
1212
    addr_list = []
1213
    myhostname = self._my_hostname
1214
    # we can skip checking whether _UnlockedGetNodeInfo returns None
1215
    # since the node list comes from _UnlocketGetNodeList, and we are
1216
    # called with the lock held, so no modifications should take place
1217
    # in between
1218
    for node_name in self._UnlockedGetNodeList():
1219
      if node_name == myhostname:
1220
        continue
1221
      node_info = self._UnlockedGetNodeInfo(node_name)
1222
      if not node_info.master_candidate:
1223
        continue
1224
      node_list.append(node_info.name)
1225
      addr_list.append(node_info.primary_ip)
1226

    
1227
    result = rpc.RpcRunner.call_upload_file(node_list, self._cfg_file,
1228
                                            address_list=addr_list)
1229
    for to_node, to_result in result.items():
1230
      msg = to_result.fail_msg
1231
      if msg:
1232
        msg = ("Copy of file %s to node %s failed: %s" %
1233
               (self._cfg_file, to_node, msg))
1234
        logging.error(msg)
1235

    
1236
        if feedback_fn:
1237
          feedback_fn(msg)
1238

    
1239
        bad = True
1240

    
1241
    return not bad
1242

    
1243
  def _WriteConfig(self, destination=None, feedback_fn=None):
1244
    """Write the configuration data to persistent storage.
1245

1246
    """
1247
    assert feedback_fn is None or callable(feedback_fn)
1248

    
1249
    # Warn on config errors, but don't abort the save - the
1250
    # configuration has already been modified, and we can't revert;
1251
    # the best we can do is to warn the user and save as is, leaving
1252
    # recovery to the user
1253
    config_errors = self._UnlockedVerifyConfig()
1254
    if config_errors:
1255
      errmsg = ("Configuration data is not consistent: %s" %
1256
                (", ".join(config_errors)))
1257
      logging.critical(errmsg)
1258
      if feedback_fn:
1259
        feedback_fn(errmsg)
1260

    
1261
    if destination is None:
1262
      destination = self._cfg_file
1263
    self._BumpSerialNo()
1264
    txt = serializer.Dump(self._config_data.ToDict())
1265

    
1266
    utils.WriteFile(destination, data=txt)
1267

    
1268
    self.write_count += 1
1269

    
1270
    # and redistribute the config file to master candidates
1271
    self._DistributeConfig(feedback_fn)
1272

    
1273
    # Write ssconf files on all nodes (including locally)
1274
    if self._last_cluster_serial < self._config_data.cluster.serial_no:
1275
      if not self._offline:
1276
        result = rpc.RpcRunner.call_write_ssconf_files(
1277
          self._UnlockedGetNodeList(),
1278
          self._UnlockedGetSsconfValues())
1279

    
1280
        for nname, nresu in result.items():
1281
          msg = nresu.fail_msg
1282
          if msg:
1283
            errmsg = ("Error while uploading ssconf files to"
1284
                      " node %s: %s" % (nname, msg))
1285
            logging.warning(errmsg)
1286

    
1287
            if feedback_fn:
1288
              feedback_fn(errmsg)
1289

    
1290
      self._last_cluster_serial = self._config_data.cluster.serial_no
1291

    
1292
  def _UnlockedGetSsconfValues(self):
1293
    """Return the values needed by ssconf.
1294

1295
    @rtype: dict
1296
    @return: a dictionary with keys the ssconf names and values their
1297
        associated value
1298

1299
    """
1300
    fn = "\n".join
1301
    instance_names = utils.NiceSort(self._UnlockedGetInstanceList())
1302
    node_names = utils.NiceSort(self._UnlockedGetNodeList())
1303
    node_info = [self._UnlockedGetNodeInfo(name) for name in node_names]
1304
    node_pri_ips = ["%s %s" % (ninfo.name, ninfo.primary_ip)
1305
                    for ninfo in node_info]
1306
    node_snd_ips = ["%s %s" % (ninfo.name, ninfo.secondary_ip)
1307
                    for ninfo in node_info]
1308

    
1309
    instance_data = fn(instance_names)
1310
    off_data = fn(node.name for node in node_info if node.offline)
1311
    on_data = fn(node.name for node in node_info if not node.offline)
1312
    mc_data = fn(node.name for node in node_info if node.master_candidate)
1313
    mc_ips_data = fn(node.primary_ip for node in node_info
1314
                     if node.master_candidate)
1315
    node_data = fn(node_names)
1316
    node_pri_ips_data = fn(node_pri_ips)
1317
    node_snd_ips_data = fn(node_snd_ips)
1318

    
1319
    cluster = self._config_data.cluster
1320
    cluster_tags = fn(cluster.GetTags())
1321
    return {
1322
      constants.SS_CLUSTER_NAME: cluster.cluster_name,
1323
      constants.SS_CLUSTER_TAGS: cluster_tags,
1324
      constants.SS_FILE_STORAGE_DIR: cluster.file_storage_dir,
1325
      constants.SS_MASTER_CANDIDATES: mc_data,
1326
      constants.SS_MASTER_CANDIDATES_IPS: mc_ips_data,
1327
      constants.SS_MASTER_IP: cluster.master_ip,
1328
      constants.SS_MASTER_NETDEV: cluster.master_netdev,
1329
      constants.SS_MASTER_NODE: cluster.master_node,
1330
      constants.SS_NODE_LIST: node_data,
1331
      constants.SS_NODE_PRIMARY_IPS: node_pri_ips_data,
1332
      constants.SS_NODE_SECONDARY_IPS: node_snd_ips_data,
1333
      constants.SS_OFFLINE_NODES: off_data,
1334
      constants.SS_ONLINE_NODES: on_data,
1335
      constants.SS_INSTANCE_LIST: instance_data,
1336
      constants.SS_RELEASE_VERSION: constants.RELEASE_VERSION,
1337
      }
1338

    
1339
  @locking.ssynchronized(_config_lock, shared=1)
1340
  def GetVGName(self):
1341
    """Return the volume group name.
1342

1343
    """
1344
    return self._config_data.cluster.volume_group_name
1345

    
1346
  @locking.ssynchronized(_config_lock)
1347
  def SetVGName(self, vg_name):
1348
    """Set the volume group name.
1349

1350
    """
1351
    self._config_data.cluster.volume_group_name = vg_name
1352
    self._config_data.cluster.serial_no += 1
1353
    self._WriteConfig()
1354

    
1355
  @locking.ssynchronized(_config_lock, shared=1)
1356
  def GetMACPrefix(self):
1357
    """Return the mac prefix.
1358

1359
    """
1360
    return self._config_data.cluster.mac_prefix
1361

    
1362
  @locking.ssynchronized(_config_lock, shared=1)
1363
  def GetClusterInfo(self):
1364
    """Returns information about the cluster
1365

1366
    @rtype: L{objects.Cluster}
1367
    @return: the cluster object
1368

1369
    """
1370
    return self._config_data.cluster
1371

    
1372
  @locking.ssynchronized(_config_lock)
1373
  def Update(self, target, feedback_fn):
1374
    """Notify function to be called after updates.
1375

1376
    This function must be called when an object (as returned by
1377
    GetInstanceInfo, GetNodeInfo, GetCluster) has been updated and the
1378
    caller wants the modifications saved to the backing store. Note
1379
    that all modified objects will be saved, but the target argument
1380
    is the one the caller wants to ensure that it's saved.
1381

1382
    @param target: an instance of either L{objects.Cluster},
1383
        L{objects.Node} or L{objects.Instance} which is existing in
1384
        the cluster
1385
    @param feedback_fn: Callable feedback function
1386

1387
    """
1388
    if self._config_data is None:
1389
      raise errors.ProgrammerError("Configuration file not read,"
1390
                                   " cannot save.")
1391
    update_serial = False
1392
    if isinstance(target, objects.Cluster):
1393
      test = target == self._config_data.cluster
1394
    elif isinstance(target, objects.Node):
1395
      test = target in self._config_data.nodes.values()
1396
      update_serial = True
1397
    elif isinstance(target, objects.Instance):
1398
      test = target in self._config_data.instances.values()
1399
    else:
1400
      raise errors.ProgrammerError("Invalid object type (%s) passed to"
1401
                                   " ConfigWriter.Update" % type(target))
1402
    if not test:
1403
      raise errors.ConfigurationError("Configuration updated since object"
1404
                                      " has been read or unknown object")
1405
    target.serial_no += 1
1406
    target.mtime = now = time.time()
1407

    
1408
    if update_serial:
1409
      # for node updates, we need to increase the cluster serial too
1410
      self._config_data.cluster.serial_no += 1
1411
      self._config_data.cluster.mtime = now
1412

    
1413
    if isinstance(target, objects.Instance):
1414
      self._UnlockedReleaseDRBDMinors(target.name)
1415

    
1416
    self._WriteConfig(feedback_fn=feedback_fn)
1417

    
1418
  @locking.ssynchronized(_config_lock)
1419
  def DropECReservations(self, ec_id):
1420
    """Drop per-execution-context reservations
1421

1422
    """
1423
    self._temporary_ids.DropECReservations(ec_id)
1424
    self._temporary_macs.DropECReservations(ec_id)
1425
    self._temporary_secrets.DropECReservations(ec_id)
1426