Statistics
| Branch: | Tag: | Revision:

root / lib / config.py @ 8785b71b

History | View | Annotate | Download (62.4 kB)

1
#
2
#
3

    
4
# Copyright (C) 2006, 2007, 2008, 2009, 2010, 2011 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
# pylint: disable-msg=R0904
35
# R0904: Too many public methods
36

    
37
import os
38
import random
39
import logging
40
import time
41

    
42
from ganeti import errors
43
from ganeti import locking
44
from ganeti import utils
45
from ganeti import constants
46
from ganeti import rpc
47
from ganeti import objects
48
from ganeti import serializer
49
from ganeti import uidpool
50
from ganeti import netutils
51
from ganeti import runtime
52

    
53

    
54
_config_lock = locking.SharedLock("ConfigWriter")
55

    
56
# job id used for resource management at config upgrade time
57
_UPGRADE_CONFIG_JID = "jid-cfg-upgrade"
58

    
59

    
60
def _ValidateConfig(data):
61
  """Verifies that a configuration objects looks valid.
62

63
  This only verifies the version of the configuration.
64

65
  @raise errors.ConfigurationError: if the version differs from what
66
      we expect
67

68
  """
69
  if data.version != constants.CONFIG_VERSION:
70
    raise errors.ConfigVersionMismatch(constants.CONFIG_VERSION, data.version)
71

    
72

    
73
class TemporaryReservationManager:
74
  """A temporary resource reservation manager.
75

76
  This is used to reserve resources in a job, before using them, making sure
77
  other jobs cannot get them in the meantime.
78

79
  """
80
  def __init__(self):
81
    self._ec_reserved = {}
82

    
83
  def Reserved(self, resource):
84
    for holder_reserved in self._ec_reserved.values():
85
      if resource in holder_reserved:
86
        return True
87
    return False
88

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

    
98
  def DropECReservations(self, ec_id):
99
    if ec_id in self._ec_reserved:
100
      del self._ec_reserved[ec_id]
101

    
102
  def GetReserved(self):
103
    all_reserved = set()
104
    for holder_reserved in self._ec_reserved.values():
105
      all_reserved.update(holder_reserved)
106
    return all_reserved
107

    
108
  def Generate(self, existing, generate_one_fn, ec_id):
109
    """Generate a new resource of this type
110

111
    """
112
    assert callable(generate_one_fn)
113

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

    
127

    
128
class ConfigWriter:
129
  """The interface to the cluster configuration.
130

131
  @ivar _temporary_lvs: reservation manager for temporary LVs
132
  @ivar _all_rms: a list of all temporary reservation managers
133

134
  """
135
  def __init__(self, cfg_file=None, offline=False, _getents=runtime.GetEnts,
136
               accept_foreign=False):
137
    self.write_count = 0
138
    self._lock = _config_lock
139
    self._config_data = None
140
    self._offline = offline
141
    if cfg_file is None:
142
      self._cfg_file = constants.CLUSTER_CONF_FILE
143
    else:
144
      self._cfg_file = cfg_file
145
    self._getents = _getents
146
    self._temporary_ids = TemporaryReservationManager()
147
    self._temporary_drbds = {}
148
    self._temporary_macs = TemporaryReservationManager()
149
    self._temporary_secrets = TemporaryReservationManager()
150
    self._temporary_lvs = TemporaryReservationManager()
151
    self._all_rms = [self._temporary_ids, self._temporary_macs,
152
                     self._temporary_secrets, self._temporary_lvs]
153
    # Note: in order to prevent errors when resolving our name in
154
    # _DistributeConfig, we compute it here once and reuse it; it's
155
    # better to raise an error before starting to modify the config
156
    # file than after it was modified
157
    self._my_hostname = netutils.Hostname.GetSysName()
158
    self._last_cluster_serial = -1
159
    self._cfg_id = None
160
    self._OpenConfig(accept_foreign)
161

    
162
  # this method needs to be static, so that we can call it on the class
163
  @staticmethod
164
  def IsCluster():
165
    """Check if the cluster is configured.
166

167
    """
168
    return os.path.exists(constants.CLUSTER_CONF_FILE)
169

    
170
  def _GenerateOneMAC(self):
171
    """Generate one mac address
172

173
    """
174
    prefix = self._config_data.cluster.mac_prefix
175
    byte1 = random.randrange(0, 256)
176
    byte2 = random.randrange(0, 256)
177
    byte3 = random.randrange(0, 256)
178
    mac = "%s:%02x:%02x:%02x" % (prefix, byte1, byte2, byte3)
179
    return mac
180

    
181
  @locking.ssynchronized(_config_lock, shared=1)
182
  def GetNdParams(self, node):
183
    """Get the node params populated with cluster defaults.
184

185
    @type node: L{object.Node}
186
    @param node: The node we want to know the params for
187
    @return: A dict with the filled in node params
188

189
    """
190
    nodegroup = self._UnlockedGetNodeGroup(node.group)
191
    return self._config_data.cluster.FillND(node, nodegroup)
192

    
193
  @locking.ssynchronized(_config_lock, shared=1)
194
  def GenerateMAC(self, ec_id):
195
    """Generate a MAC for an instance.
196

197
    This should check the current instances for duplicates.
198

199
    """
200
    existing = self._AllMACs()
201
    return self._temporary_ids.Generate(existing, self._GenerateOneMAC, ec_id)
202

    
203
  @locking.ssynchronized(_config_lock, shared=1)
204
  def ReserveMAC(self, mac, ec_id):
205
    """Reserve a MAC for an instance.
206

207
    This only checks instances managed by this cluster, it does not
208
    check for potential collisions elsewhere.
209

210
    """
211
    all_macs = self._AllMACs()
212
    if mac in all_macs:
213
      raise errors.ReservationError("mac already in use")
214
    else:
215
      self._temporary_macs.Reserve(ec_id, mac)
216

    
217
  @locking.ssynchronized(_config_lock, shared=1)
218
  def ReserveLV(self, lv_name, ec_id):
219
    """Reserve an VG/LV pair for an instance.
220

221
    @type lv_name: string
222
    @param lv_name: the logical volume name to reserve
223

224
    """
225
    all_lvs = self._AllLVs()
226
    if lv_name in all_lvs:
227
      raise errors.ReservationError("LV already in use")
228
    else:
229
      self._temporary_lvs.Reserve(ec_id, lv_name)
230

    
231
  @locking.ssynchronized(_config_lock, shared=1)
232
  def GenerateDRBDSecret(self, ec_id):
233
    """Generate a DRBD secret.
234

235
    This checks the current disks for duplicates.
236

237
    """
238
    return self._temporary_secrets.Generate(self._AllDRBDSecrets(),
239
                                            utils.GenerateSecret,
240
                                            ec_id)
241

    
242
  def _AllLVs(self):
243
    """Compute the list of all LVs.
244

245
    """
246
    lvnames = set()
247
    for instance in self._config_data.instances.values():
248
      node_data = instance.MapLVsByNode()
249
      for lv_list in node_data.values():
250
        lvnames.update(lv_list)
251
    return lvnames
252

    
253
  def _AllIDs(self, include_temporary):
254
    """Compute the list of all UUIDs and names we have.
255

256
    @type include_temporary: boolean
257
    @param include_temporary: whether to include the _temporary_ids set
258
    @rtype: set
259
    @return: a set of IDs
260

261
    """
262
    existing = set()
263
    if include_temporary:
264
      existing.update(self._temporary_ids.GetReserved())
265
    existing.update(self._AllLVs())
266
    existing.update(self._config_data.instances.keys())
267
    existing.update(self._config_data.nodes.keys())
268
    existing.update([i.uuid for i in self._AllUUIDObjects() if i.uuid])
269
    return existing
270

    
271
  def _GenerateUniqueID(self, ec_id):
272
    """Generate an unique UUID.
273

274
    This checks the current node, instances and disk names for
275
    duplicates.
276

277
    @rtype: string
278
    @return: the unique id
279

280
    """
281
    existing = self._AllIDs(include_temporary=False)
282
    return self._temporary_ids.Generate(existing, utils.NewUUID, ec_id)
283

    
284
  @locking.ssynchronized(_config_lock, shared=1)
285
  def GenerateUniqueID(self, ec_id):
286
    """Generate an unique ID.
287

288
    This is just a wrapper over the unlocked version.
289

290
    @type ec_id: string
291
    @param ec_id: unique id for the job to reserve the id to
292

293
    """
294
    return self._GenerateUniqueID(ec_id)
295

    
296
  def _AllMACs(self):
297
    """Return all MACs present in the config.
298

299
    @rtype: list
300
    @return: the list of all MACs
301

302
    """
303
    result = []
304
    for instance in self._config_data.instances.values():
305
      for nic in instance.nics:
306
        result.append(nic.mac)
307

    
308
    return result
309

    
310
  def _AllDRBDSecrets(self):
311
    """Return all DRBD secrets present in the config.
312

313
    @rtype: list
314
    @return: the list of all DRBD secrets
315

316
    """
317
    def helper(disk, result):
318
      """Recursively gather secrets from this disk."""
319
      if disk.dev_type == constants.DT_DRBD8:
320
        result.append(disk.logical_id[5])
321
      if disk.children:
322
        for child in disk.children:
323
          helper(child, result)
324

    
325
    result = []
326
    for instance in self._config_data.instances.values():
327
      for disk in instance.disks:
328
        helper(disk, result)
329

    
330
    return result
331

    
332
  def _CheckDiskIDs(self, disk, l_ids, p_ids):
333
    """Compute duplicate disk IDs
334

335
    @type disk: L{objects.Disk}
336
    @param disk: the disk at which to start searching
337
    @type l_ids: list
338
    @param l_ids: list of current logical ids
339
    @type p_ids: list
340
    @param p_ids: list of current physical ids
341
    @rtype: list
342
    @return: a list of error messages
343

344
    """
345
    result = []
346
    if disk.logical_id is not None:
347
      if disk.logical_id in l_ids:
348
        result.append("duplicate logical id %s" % str(disk.logical_id))
349
      else:
350
        l_ids.append(disk.logical_id)
351
    if disk.physical_id is not None:
352
      if disk.physical_id in p_ids:
353
        result.append("duplicate physical id %s" % str(disk.physical_id))
354
      else:
355
        p_ids.append(disk.physical_id)
356

    
357
    if disk.children:
358
      for child in disk.children:
359
        result.extend(self._CheckDiskIDs(child, l_ids, p_ids))
360
    return result
361

    
362
  def _UnlockedVerifyConfig(self):
363
    """Verify function.
364

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

369
    """
370
    # pylint: disable-msg=R0914
371
    result = []
372
    seen_macs = []
373
    ports = {}
374
    data = self._config_data
375
    cluster = data.cluster
376
    seen_lids = []
377
    seen_pids = []
378

    
379
    # global cluster checks
380
    if not cluster.enabled_hypervisors:
381
      result.append("enabled hypervisors list doesn't have any entries")
382
    invalid_hvs = set(cluster.enabled_hypervisors) - constants.HYPER_TYPES
383
    if invalid_hvs:
384
      result.append("enabled hypervisors contains invalid entries: %s" %
385
                    invalid_hvs)
386
    missing_hvp = (set(cluster.enabled_hypervisors) -
387
                   set(cluster.hvparams.keys()))
388
    if missing_hvp:
389
      result.append("hypervisor parameters missing for the enabled"
390
                    " hypervisor(s) %s" % utils.CommaJoin(missing_hvp))
391

    
392
    if cluster.master_node not in data.nodes:
393
      result.append("cluster has invalid primary node '%s'" %
394
                    cluster.master_node)
395

    
396
    def _helper(owner, attr, value, template):
397
      try:
398
        utils.ForceDictType(value, template)
399
      except errors.GenericError, err:
400
        result.append("%s has invalid %s: %s" % (owner, attr, err))
401

    
402
    def _helper_nic(owner, params):
403
      try:
404
        objects.NIC.CheckParameterSyntax(params)
405
      except errors.ConfigurationError, err:
406
        result.append("%s has invalid nicparams: %s" % (owner, err))
407

    
408
    # check cluster parameters
409
    _helper("cluster", "beparams", cluster.SimpleFillBE({}),
410
            constants.BES_PARAMETER_TYPES)
411
    _helper("cluster", "nicparams", cluster.SimpleFillNIC({}),
412
            constants.NICS_PARAMETER_TYPES)
413
    _helper_nic("cluster", cluster.SimpleFillNIC({}))
414
    _helper("cluster", "ndparams", cluster.SimpleFillND({}),
415
            constants.NDS_PARAMETER_TYPES)
416

    
417
    # per-instance checks
418
    for instance_name in data.instances:
419
      instance = data.instances[instance_name]
420
      if instance.name != instance_name:
421
        result.append("instance '%s' is indexed by wrong name '%s'" %
422
                      (instance.name, instance_name))
423
      if instance.primary_node not in data.nodes:
424
        result.append("instance '%s' has invalid primary node '%s'" %
425
                      (instance_name, instance.primary_node))
426
      for snode in instance.secondary_nodes:
427
        if snode not in data.nodes:
428
          result.append("instance '%s' has invalid secondary node '%s'" %
429
                        (instance_name, snode))
430
      for idx, nic in enumerate(instance.nics):
431
        if nic.mac in seen_macs:
432
          result.append("instance '%s' has NIC %d mac %s duplicate" %
433
                        (instance_name, idx, nic.mac))
434
        else:
435
          seen_macs.append(nic.mac)
436
        if nic.nicparams:
437
          filled = cluster.SimpleFillNIC(nic.nicparams)
438
          owner = "instance %s nic %d" % (instance.name, idx)
439
          _helper(owner, "nicparams",
440
                  filled, constants.NICS_PARAMETER_TYPES)
441
          _helper_nic(owner, filled)
442

    
443
      # parameter checks
444
      if instance.beparams:
445
        _helper("instance %s" % instance.name, "beparams",
446
                cluster.FillBE(instance), constants.BES_PARAMETER_TYPES)
447

    
448
      # gather the drbd ports for duplicate checks
449
      for dsk in instance.disks:
450
        if dsk.dev_type in constants.LDS_DRBD:
451
          tcp_port = dsk.logical_id[2]
452
          if tcp_port not in ports:
453
            ports[tcp_port] = []
454
          ports[tcp_port].append((instance.name, "drbd disk %s" % dsk.iv_name))
455
      # gather network port reservation
456
      net_port = getattr(instance, "network_port", None)
457
      if net_port is not None:
458
        if net_port not in ports:
459
          ports[net_port] = []
460
        ports[net_port].append((instance.name, "network port"))
461

    
462
      # instance disk verify
463
      for idx, disk in enumerate(instance.disks):
464
        result.extend(["instance '%s' disk %d error: %s" %
465
                       (instance.name, idx, msg) for msg in disk.Verify()])
466
        result.extend(self._CheckDiskIDs(disk, seen_lids, seen_pids))
467

    
468
    # cluster-wide pool of free ports
469
    for free_port in cluster.tcpudp_port_pool:
470
      if free_port not in ports:
471
        ports[free_port] = []
472
      ports[free_port].append(("cluster", "port marked as free"))
473

    
474
    # compute tcp/udp duplicate ports
475
    keys = ports.keys()
476
    keys.sort()
477
    for pnum in keys:
478
      pdata = ports[pnum]
479
      if len(pdata) > 1:
480
        txt = utils.CommaJoin(["%s/%s" % val for val in pdata])
481
        result.append("tcp/udp port %s has duplicates: %s" % (pnum, txt))
482

    
483
    # highest used tcp port check
484
    if keys:
485
      if keys[-1] > cluster.highest_used_port:
486
        result.append("Highest used port mismatch, saved %s, computed %s" %
487
                      (cluster.highest_used_port, keys[-1]))
488

    
489
    if not data.nodes[cluster.master_node].master_candidate:
490
      result.append("Master node is not a master candidate")
491

    
492
    # master candidate checks
493
    mc_now, mc_max, _ = self._UnlockedGetMasterCandidateStats()
494
    if mc_now < mc_max:
495
      result.append("Not enough master candidates: actual %d, target %d" %
496
                    (mc_now, mc_max))
497

    
498
    # node checks
499
    for node_name, node in data.nodes.items():
500
      if node.name != node_name:
501
        result.append("Node '%s' is indexed by wrong name '%s'" %
502
                      (node.name, node_name))
503
      if [node.master_candidate, node.drained, node.offline].count(True) > 1:
504
        result.append("Node %s state is invalid: master_candidate=%s,"
505
                      " drain=%s, offline=%s" %
506
                      (node.name, node.master_candidate, node.drained,
507
                       node.offline))
508
      if node.group not in data.nodegroups:
509
        result.append("Node '%s' has invalid group '%s'" %
510
                      (node.name, node.group))
511
      else:
512
        _helper("node %s" % node.name, "ndparams",
513
                cluster.FillND(node, data.nodegroups[node.group]),
514
                constants.NDS_PARAMETER_TYPES)
515

    
516
    # nodegroups checks
517
    nodegroups_names = set()
518
    for nodegroup_uuid in data.nodegroups:
519
      nodegroup = data.nodegroups[nodegroup_uuid]
520
      if nodegroup.uuid != nodegroup_uuid:
521
        result.append("node group '%s' (uuid: '%s') indexed by wrong uuid '%s'"
522
                      % (nodegroup.name, nodegroup.uuid, nodegroup_uuid))
523
      if utils.UUID_RE.match(nodegroup.name.lower()):
524
        result.append("node group '%s' (uuid: '%s') has uuid-like name" %
525
                      (nodegroup.name, nodegroup.uuid))
526
      if nodegroup.name in nodegroups_names:
527
        result.append("duplicate node group name '%s'" % nodegroup.name)
528
      else:
529
        nodegroups_names.add(nodegroup.name)
530
      if nodegroup.ndparams:
531
        _helper("group %s" % nodegroup.name, "ndparams",
532
                cluster.SimpleFillND(nodegroup.ndparams),
533
                constants.NDS_PARAMETER_TYPES)
534

    
535

    
536
    # drbd minors check
537
    _, duplicates = self._UnlockedComputeDRBDMap()
538
    for node, minor, instance_a, instance_b in duplicates:
539
      result.append("DRBD minor %d on node %s is assigned twice to instances"
540
                    " %s and %s" % (minor, node, instance_a, instance_b))
541

    
542
    # IP checks
543
    default_nicparams = cluster.nicparams[constants.PP_DEFAULT]
544
    ips = {}
545

    
546
    def _AddIpAddress(ip, name):
547
      ips.setdefault(ip, []).append(name)
548

    
549
    _AddIpAddress(cluster.master_ip, "cluster_ip")
550

    
551
    for node in data.nodes.values():
552
      _AddIpAddress(node.primary_ip, "node:%s/primary" % node.name)
553
      if node.secondary_ip != node.primary_ip:
554
        _AddIpAddress(node.secondary_ip, "node:%s/secondary" % node.name)
555

    
556
    for instance in data.instances.values():
557
      for idx, nic in enumerate(instance.nics):
558
        if nic.ip is None:
559
          continue
560

    
561
        nicparams = objects.FillDict(default_nicparams, nic.nicparams)
562
        nic_mode = nicparams[constants.NIC_MODE]
563
        nic_link = nicparams[constants.NIC_LINK]
564

    
565
        if nic_mode == constants.NIC_MODE_BRIDGED:
566
          link = "bridge:%s" % nic_link
567
        elif nic_mode == constants.NIC_MODE_ROUTED:
568
          link = "route:%s" % nic_link
569
        else:
570
          raise errors.ProgrammerError("NIC mode '%s' not handled" % nic_mode)
571

    
572
        _AddIpAddress("%s/%s" % (link, nic.ip),
573
                      "instance:%s/nic:%d" % (instance.name, idx))
574

    
575
    for ip, owners in ips.items():
576
      if len(owners) > 1:
577
        result.append("IP address %s is used by multiple owners: %s" %
578
                      (ip, utils.CommaJoin(owners)))
579

    
580
    return result
581

    
582
  @locking.ssynchronized(_config_lock, shared=1)
583
  def VerifyConfig(self):
584
    """Verify function.
585

586
    This is just a wrapper over L{_UnlockedVerifyConfig}.
587

588
    @rtype: list
589
    @return: a list of error messages; a non-empty list signifies
590
        configuration errors
591

592
    """
593
    return self._UnlockedVerifyConfig()
594

    
595
  def _UnlockedSetDiskID(self, disk, node_name):
596
    """Convert the unique ID to the ID needed on the target nodes.
597

598
    This is used only for drbd, which needs ip/port configuration.
599

600
    The routine descends down and updates its children also, because
601
    this helps when the only the top device is passed to the remote
602
    node.
603

604
    This function is for internal use, when the config lock is already held.
605

606
    """
607
    if disk.children:
608
      for child in disk.children:
609
        self._UnlockedSetDiskID(child, node_name)
610

    
611
    if disk.logical_id is None and disk.physical_id is not None:
612
      return
613
    if disk.dev_type == constants.LD_DRBD8:
614
      pnode, snode, port, pminor, sminor, secret = disk.logical_id
615
      if node_name not in (pnode, snode):
616
        raise errors.ConfigurationError("DRBD device not knowing node %s" %
617
                                        node_name)
618
      pnode_info = self._UnlockedGetNodeInfo(pnode)
619
      snode_info = self._UnlockedGetNodeInfo(snode)
620
      if pnode_info is None or snode_info is None:
621
        raise errors.ConfigurationError("Can't find primary or secondary node"
622
                                        " for %s" % str(disk))
623
      p_data = (pnode_info.secondary_ip, port)
624
      s_data = (snode_info.secondary_ip, port)
625
      if pnode == node_name:
626
        disk.physical_id = p_data + s_data + (pminor, secret)
627
      else: # it must be secondary, we tested above
628
        disk.physical_id = s_data + p_data + (sminor, secret)
629
    else:
630
      disk.physical_id = disk.logical_id
631
    return
632

    
633
  @locking.ssynchronized(_config_lock)
634
  def SetDiskID(self, disk, node_name):
635
    """Convert the unique ID to the ID needed on the target nodes.
636

637
    This is used only for drbd, which needs ip/port configuration.
638

639
    The routine descends down and updates its children also, because
640
    this helps when the only the top device is passed to the remote
641
    node.
642

643
    """
644
    return self._UnlockedSetDiskID(disk, node_name)
645

    
646
  @locking.ssynchronized(_config_lock)
647
  def AddTcpUdpPort(self, port):
648
    """Adds a new port to the available port pool.
649

650
    """
651
    if not isinstance(port, int):
652
      raise errors.ProgrammerError("Invalid type passed for port")
653

    
654
    self._config_data.cluster.tcpudp_port_pool.add(port)
655
    self._WriteConfig()
656

    
657
  @locking.ssynchronized(_config_lock, shared=1)
658
  def GetPortList(self):
659
    """Returns a copy of the current port list.
660

661
    """
662
    return self._config_data.cluster.tcpudp_port_pool.copy()
663

    
664
  @locking.ssynchronized(_config_lock)
665
  def AllocatePort(self):
666
    """Allocate a port.
667

668
    The port will be taken from the available port pool or from the
669
    default port range (and in this case we increase
670
    highest_used_port).
671

672
    """
673
    # If there are TCP/IP ports configured, we use them first.
674
    if self._config_data.cluster.tcpudp_port_pool:
675
      port = self._config_data.cluster.tcpudp_port_pool.pop()
676
    else:
677
      port = self._config_data.cluster.highest_used_port + 1
678
      if port >= constants.LAST_DRBD_PORT:
679
        raise errors.ConfigurationError("The highest used port is greater"
680
                                        " than %s. Aborting." %
681
                                        constants.LAST_DRBD_PORT)
682
      self._config_data.cluster.highest_used_port = port
683

    
684
    self._WriteConfig()
685
    return port
686

    
687
  def _UnlockedComputeDRBDMap(self):
688
    """Compute the used DRBD minor/nodes.
689

690
    @rtype: (dict, list)
691
    @return: dictionary of node_name: dict of minor: instance_name;
692
        the returned dict will have all the nodes in it (even if with
693
        an empty list), and a list of duplicates; if the duplicates
694
        list is not empty, the configuration is corrupted and its caller
695
        should raise an exception
696

697
    """
698
    def _AppendUsedPorts(instance_name, disk, used):
699
      duplicates = []
700
      if disk.dev_type == constants.LD_DRBD8 and len(disk.logical_id) >= 5:
701
        node_a, node_b, _, minor_a, minor_b = disk.logical_id[:5]
702
        for node, port in ((node_a, minor_a), (node_b, minor_b)):
703
          assert node in used, ("Node '%s' of instance '%s' not found"
704
                                " in node list" % (node, instance_name))
705
          if port in used[node]:
706
            duplicates.append((node, port, instance_name, used[node][port]))
707
          else:
708
            used[node][port] = instance_name
709
      if disk.children:
710
        for child in disk.children:
711
          duplicates.extend(_AppendUsedPorts(instance_name, child, used))
712
      return duplicates
713

    
714
    duplicates = []
715
    my_dict = dict((node, {}) for node in self._config_data.nodes)
716
    for instance in self._config_data.instances.itervalues():
717
      for disk in instance.disks:
718
        duplicates.extend(_AppendUsedPorts(instance.name, disk, my_dict))
719
    for (node, minor), instance in self._temporary_drbds.iteritems():
720
      if minor in my_dict[node] and my_dict[node][minor] != instance:
721
        duplicates.append((node, minor, instance, my_dict[node][minor]))
722
      else:
723
        my_dict[node][minor] = instance
724
    return my_dict, duplicates
725

    
726
  @locking.ssynchronized(_config_lock)
727
  def ComputeDRBDMap(self):
728
    """Compute the used DRBD minor/nodes.
729

730
    This is just a wrapper over L{_UnlockedComputeDRBDMap}.
731

732
    @return: dictionary of node_name: dict of minor: instance_name;
733
        the returned dict will have all the nodes in it (even if with
734
        an empty list).
735

736
    """
737
    d_map, duplicates = self._UnlockedComputeDRBDMap()
738
    if duplicates:
739
      raise errors.ConfigurationError("Duplicate DRBD ports detected: %s" %
740
                                      str(duplicates))
741
    return d_map
742

    
743
  @locking.ssynchronized(_config_lock)
744
  def AllocateDRBDMinor(self, nodes, instance):
745
    """Allocate a drbd minor.
746

747
    The free minor will be automatically computed from the existing
748
    devices. A node can be given multiple times in order to allocate
749
    multiple minors. The result is the list of minors, in the same
750
    order as the passed nodes.
751

752
    @type instance: string
753
    @param instance: the instance for which we allocate minors
754

755
    """
756
    assert isinstance(instance, basestring), \
757
           "Invalid argument '%s' passed to AllocateDRBDMinor" % instance
758

    
759
    d_map, duplicates = self._UnlockedComputeDRBDMap()
760
    if duplicates:
761
      raise errors.ConfigurationError("Duplicate DRBD ports detected: %s" %
762
                                      str(duplicates))
763
    result = []
764
    for nname in nodes:
765
      ndata = d_map[nname]
766
      if not ndata:
767
        # no minors used, we can start at 0
768
        result.append(0)
769
        ndata[0] = instance
770
        self._temporary_drbds[(nname, 0)] = instance
771
        continue
772
      keys = ndata.keys()
773
      keys.sort()
774
      ffree = utils.FirstFree(keys)
775
      if ffree is None:
776
        # return the next minor
777
        # TODO: implement high-limit check
778
        minor = keys[-1] + 1
779
      else:
780
        minor = ffree
781
      # double-check minor against current instances
782
      assert minor not in d_map[nname], \
783
             ("Attempt to reuse allocated DRBD minor %d on node %s,"
784
              " already allocated to instance %s" %
785
              (minor, nname, d_map[nname][minor]))
786
      ndata[minor] = instance
787
      # double-check minor against reservation
788
      r_key = (nname, minor)
789
      assert r_key not in self._temporary_drbds, \
790
             ("Attempt to reuse reserved DRBD minor %d on node %s,"
791
              " reserved for instance %s" %
792
              (minor, nname, self._temporary_drbds[r_key]))
793
      self._temporary_drbds[r_key] = instance
794
      result.append(minor)
795
    logging.debug("Request to allocate drbd minors, input: %s, returning %s",
796
                  nodes, result)
797
    return result
798

    
799
  def _UnlockedReleaseDRBDMinors(self, instance):
800
    """Release temporary drbd minors allocated for a given instance.
801

802
    @type instance: string
803
    @param instance: the instance for which temporary minors should be
804
                     released
805

806
    """
807
    assert isinstance(instance, basestring), \
808
           "Invalid argument passed to ReleaseDRBDMinors"
809
    for key, name in self._temporary_drbds.items():
810
      if name == instance:
811
        del self._temporary_drbds[key]
812

    
813
  @locking.ssynchronized(_config_lock)
814
  def ReleaseDRBDMinors(self, instance):
815
    """Release temporary drbd minors allocated for a given instance.
816

817
    This should be called on the error paths, on the success paths
818
    it's automatically called by the ConfigWriter add and update
819
    functions.
820

821
    This function is just a wrapper over L{_UnlockedReleaseDRBDMinors}.
822

823
    @type instance: string
824
    @param instance: the instance for which temporary minors should be
825
                     released
826

827
    """
828
    self._UnlockedReleaseDRBDMinors(instance)
829

    
830
  @locking.ssynchronized(_config_lock, shared=1)
831
  def GetConfigVersion(self):
832
    """Get the configuration version.
833

834
    @return: Config version
835

836
    """
837
    return self._config_data.version
838

    
839
  @locking.ssynchronized(_config_lock, shared=1)
840
  def GetClusterName(self):
841
    """Get cluster name.
842

843
    @return: Cluster name
844

845
    """
846
    return self._config_data.cluster.cluster_name
847

    
848
  @locking.ssynchronized(_config_lock, shared=1)
849
  def GetMasterNode(self):
850
    """Get the hostname of the master node for this cluster.
851

852
    @return: Master hostname
853

854
    """
855
    return self._config_data.cluster.master_node
856

    
857
  @locking.ssynchronized(_config_lock, shared=1)
858
  def GetMasterIP(self):
859
    """Get the IP of the master node for this cluster.
860

861
    @return: Master IP
862

863
    """
864
    return self._config_data.cluster.master_ip
865

    
866
  @locking.ssynchronized(_config_lock, shared=1)
867
  def GetMasterNetdev(self):
868
    """Get the master network device for this cluster.
869

870
    """
871
    return self._config_data.cluster.master_netdev
872

    
873
  @locking.ssynchronized(_config_lock, shared=1)
874
  def GetFileStorageDir(self):
875
    """Get the file storage dir for this cluster.
876

877
    """
878
    return self._config_data.cluster.file_storage_dir
879

    
880
  @locking.ssynchronized(_config_lock, shared=1)
881
  def GetHypervisorType(self):
882
    """Get the hypervisor type for this cluster.
883

884
    """
885
    return self._config_data.cluster.enabled_hypervisors[0]
886

    
887
  @locking.ssynchronized(_config_lock, shared=1)
888
  def GetHostKey(self):
889
    """Return the rsa hostkey from the config.
890

891
    @rtype: string
892
    @return: the rsa hostkey
893

894
    """
895
    return self._config_data.cluster.rsahostkeypub
896

    
897
  @locking.ssynchronized(_config_lock, shared=1)
898
  def GetDefaultIAllocator(self):
899
    """Get the default instance allocator for this cluster.
900

901
    """
902
    return self._config_data.cluster.default_iallocator
903

    
904
  @locking.ssynchronized(_config_lock, shared=1)
905
  def GetPrimaryIPFamily(self):
906
    """Get cluster primary ip family.
907

908
    @return: primary ip family
909

910
    """
911
    return self._config_data.cluster.primary_ip_family
912

    
913
  @locking.ssynchronized(_config_lock)
914
  def AddNodeGroup(self, group, ec_id, check_uuid=True):
915
    """Add a node group to the configuration.
916

917
    This method calls group.UpgradeConfig() to fill any missing attributes
918
    according to their default values.
919

920
    @type group: L{objects.NodeGroup}
921
    @param group: the NodeGroup object to add
922
    @type ec_id: string
923
    @param ec_id: unique id for the job to use when creating a missing UUID
924
    @type check_uuid: bool
925
    @param check_uuid: add an UUID to the group if it doesn't have one or, if
926
                       it does, ensure that it does not exist in the
927
                       configuration already
928

929
    """
930
    self._UnlockedAddNodeGroup(group, ec_id, check_uuid)
931
    self._WriteConfig()
932

    
933
  def _UnlockedAddNodeGroup(self, group, ec_id, check_uuid):
934
    """Add a node group to the configuration.
935

936
    """
937
    logging.info("Adding node group %s to configuration", group.name)
938

    
939
    # Some code might need to add a node group with a pre-populated UUID
940
    # generated with ConfigWriter.GenerateUniqueID(). We allow them to bypass
941
    # the "does this UUID" exist already check.
942
    if check_uuid:
943
      self._EnsureUUID(group, ec_id)
944

    
945
    try:
946
      existing_uuid = self._UnlockedLookupNodeGroup(group.name)
947
    except errors.OpPrereqError:
948
      pass
949
    else:
950
      raise errors.OpPrereqError("Desired group name '%s' already exists as a"
951
                                 " node group (UUID: %s)" %
952
                                 (group.name, existing_uuid),
953
                                 errors.ECODE_EXISTS)
954

    
955
    group.serial_no = 1
956
    group.ctime = group.mtime = time.time()
957
    group.UpgradeConfig()
958

    
959
    self._config_data.nodegroups[group.uuid] = group
960
    self._config_data.cluster.serial_no += 1
961

    
962
  @locking.ssynchronized(_config_lock)
963
  def RemoveNodeGroup(self, group_uuid):
964
    """Remove a node group from the configuration.
965

966
    @type group_uuid: string
967
    @param group_uuid: the UUID of the node group to remove
968

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

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

    
975
    assert len(self._config_data.nodegroups) != 1, \
976
            "Group '%s' is the only group, cannot be removed" % group_uuid
977

    
978
    del self._config_data.nodegroups[group_uuid]
979
    self._config_data.cluster.serial_no += 1
980
    self._WriteConfig()
981

    
982
  def _UnlockedLookupNodeGroup(self, target):
983
    """Lookup a node group's UUID.
984

985
    @type target: string or None
986
    @param target: group name or UUID or None to look for the default
987
    @rtype: string
988
    @return: nodegroup UUID
989
    @raises errors.OpPrereqError: when the target group cannot be found
990

991
    """
992
    if target is None:
993
      if len(self._config_data.nodegroups) != 1:
994
        raise errors.OpPrereqError("More than one node group exists. Target"
995
                                   " group must be specified explicitely.")
996
      else:
997
        return self._config_data.nodegroups.keys()[0]
998
    if target in self._config_data.nodegroups:
999
      return target
1000
    for nodegroup in self._config_data.nodegroups.values():
1001
      if nodegroup.name == target:
1002
        return nodegroup.uuid
1003
    raise errors.OpPrereqError("Node group '%s' not found" % target,
1004
                               errors.ECODE_NOENT)
1005

    
1006
  @locking.ssynchronized(_config_lock, shared=1)
1007
  def LookupNodeGroup(self, target):
1008
    """Lookup a node group's UUID.
1009

1010
    This function is just a wrapper over L{_UnlockedLookupNodeGroup}.
1011

1012
    @type target: string or None
1013
    @param target: group name or UUID or None to look for the default
1014
    @rtype: string
1015
    @return: nodegroup UUID
1016

1017
    """
1018
    return self._UnlockedLookupNodeGroup(target)
1019

    
1020
  def _UnlockedGetNodeGroup(self, uuid):
1021
    """Lookup a node group.
1022

1023
    @type uuid: string
1024
    @param uuid: group UUID
1025
    @rtype: L{objects.NodeGroup} or None
1026
    @return: nodegroup object, or None if not found
1027

1028
    """
1029
    if uuid not in self._config_data.nodegroups:
1030
      return None
1031

    
1032
    return self._config_data.nodegroups[uuid]
1033

    
1034
  @locking.ssynchronized(_config_lock, shared=1)
1035
  def GetNodeGroup(self, uuid):
1036
    """Lookup a node group.
1037

1038
    @type uuid: string
1039
    @param uuid: group UUID
1040
    @rtype: L{objects.NodeGroup} or None
1041
    @return: nodegroup object, or None if not found
1042

1043
    """
1044
    return self._UnlockedGetNodeGroup(uuid)
1045

    
1046
  @locking.ssynchronized(_config_lock, shared=1)
1047
  def GetAllNodeGroupsInfo(self):
1048
    """Get the configuration of all node groups.
1049

1050
    """
1051
    return dict(self._config_data.nodegroups)
1052

    
1053
  @locking.ssynchronized(_config_lock, shared=1)
1054
  def GetNodeGroupList(self):
1055
    """Get a list of node groups.
1056

1057
    """
1058
    return self._config_data.nodegroups.keys()
1059

    
1060
  @locking.ssynchronized(_config_lock)
1061
  def AddInstance(self, instance, ec_id):
1062
    """Add an instance to the config.
1063

1064
    This should be used after creating a new instance.
1065

1066
    @type instance: L{objects.Instance}
1067
    @param instance: the instance object
1068

1069
    """
1070
    if not isinstance(instance, objects.Instance):
1071
      raise errors.ProgrammerError("Invalid type passed to AddInstance")
1072

    
1073
    if instance.disk_template != constants.DT_DISKLESS:
1074
      all_lvs = instance.MapLVsByNode()
1075
      logging.info("Instance '%s' DISK_LAYOUT: %s", instance.name, all_lvs)
1076

    
1077
    all_macs = self._AllMACs()
1078
    for nic in instance.nics:
1079
      if nic.mac in all_macs:
1080
        raise errors.ConfigurationError("Cannot add instance %s:"
1081
                                        " MAC address '%s' already in use." %
1082
                                        (instance.name, nic.mac))
1083

    
1084
    self._EnsureUUID(instance, ec_id)
1085

    
1086
    instance.serial_no = 1
1087
    instance.ctime = instance.mtime = time.time()
1088
    self._config_data.instances[instance.name] = instance
1089
    self._config_data.cluster.serial_no += 1
1090
    self._UnlockedReleaseDRBDMinors(instance.name)
1091
    self._WriteConfig()
1092

    
1093
  def _EnsureUUID(self, item, ec_id):
1094
    """Ensures a given object has a valid UUID.
1095

1096
    @param item: the instance or node to be checked
1097
    @param ec_id: the execution context id for the uuid reservation
1098

1099
    """
1100
    if not item.uuid:
1101
      item.uuid = self._GenerateUniqueID(ec_id)
1102
    elif item.uuid in self._AllIDs(include_temporary=True):
1103
      raise errors.ConfigurationError("Cannot add '%s': UUID %s already"
1104
                                      " in use" % (item.name, item.uuid))
1105

    
1106
  def _SetInstanceStatus(self, instance_name, status):
1107
    """Set the instance's status to a given value.
1108

1109
    """
1110
    assert isinstance(status, bool), \
1111
           "Invalid status '%s' passed to SetInstanceStatus" % (status,)
1112

    
1113
    if instance_name not in self._config_data.instances:
1114
      raise errors.ConfigurationError("Unknown instance '%s'" %
1115
                                      instance_name)
1116
    instance = self._config_data.instances[instance_name]
1117
    if instance.admin_up != status:
1118
      instance.admin_up = status
1119
      instance.serial_no += 1
1120
      instance.mtime = time.time()
1121
      self._WriteConfig()
1122

    
1123
  @locking.ssynchronized(_config_lock)
1124
  def MarkInstanceUp(self, instance_name):
1125
    """Mark the instance status to up in the config.
1126

1127
    """
1128
    self._SetInstanceStatus(instance_name, True)
1129

    
1130
  @locking.ssynchronized(_config_lock)
1131
  def RemoveInstance(self, instance_name):
1132
    """Remove the instance from the configuration.
1133

1134
    """
1135
    if instance_name not in self._config_data.instances:
1136
      raise errors.ConfigurationError("Unknown instance '%s'" % instance_name)
1137
    del self._config_data.instances[instance_name]
1138
    self._config_data.cluster.serial_no += 1
1139
    self._WriteConfig()
1140

    
1141
  @locking.ssynchronized(_config_lock)
1142
  def RenameInstance(self, old_name, new_name):
1143
    """Rename an instance.
1144

1145
    This needs to be done in ConfigWriter and not by RemoveInstance
1146
    combined with AddInstance as only we can guarantee an atomic
1147
    rename.
1148

1149
    """
1150
    if old_name not in self._config_data.instances:
1151
      raise errors.ConfigurationError("Unknown instance '%s'" % old_name)
1152
    inst = self._config_data.instances[old_name]
1153
    del self._config_data.instances[old_name]
1154
    inst.name = new_name
1155

    
1156
    for disk in inst.disks:
1157
      if disk.dev_type == constants.LD_FILE:
1158
        # rename the file paths in logical and physical id
1159
        file_storage_dir = os.path.dirname(os.path.dirname(disk.logical_id[1]))
1160
        disk_fname = "disk%s" % disk.iv_name.split("/")[1]
1161
        disk.physical_id = disk.logical_id = (disk.logical_id[0],
1162
                                              utils.PathJoin(file_storage_dir,
1163
                                                             inst.name,
1164
                                                             disk_fname))
1165

    
1166
    # Force update of ssconf files
1167
    self._config_data.cluster.serial_no += 1
1168

    
1169
    self._config_data.instances[inst.name] = inst
1170
    self._WriteConfig()
1171

    
1172
  @locking.ssynchronized(_config_lock)
1173
  def MarkInstanceDown(self, instance_name):
1174
    """Mark the status of an instance to down in the configuration.
1175

1176
    """
1177
    self._SetInstanceStatus(instance_name, False)
1178

    
1179
  def _UnlockedGetInstanceList(self):
1180
    """Get the list of instances.
1181

1182
    This function is for internal use, when the config lock is already held.
1183

1184
    """
1185
    return self._config_data.instances.keys()
1186

    
1187
  @locking.ssynchronized(_config_lock, shared=1)
1188
  def GetInstanceList(self):
1189
    """Get the list of instances.
1190

1191
    @return: array of instances, ex. ['instance2.example.com',
1192
        'instance1.example.com']
1193

1194
    """
1195
    return self._UnlockedGetInstanceList()
1196

    
1197
  @locking.ssynchronized(_config_lock, shared=1)
1198
  def ExpandInstanceName(self, short_name):
1199
    """Attempt to expand an incomplete instance name.
1200

1201
    """
1202
    return utils.MatchNameComponent(short_name,
1203
                                    self._config_data.instances.keys(),
1204
                                    case_sensitive=False)
1205

    
1206
  def _UnlockedGetInstanceInfo(self, instance_name):
1207
    """Returns information about an instance.
1208

1209
    This function is for internal use, when the config lock is already held.
1210

1211
    """
1212
    if instance_name not in self._config_data.instances:
1213
      return None
1214

    
1215
    return self._config_data.instances[instance_name]
1216

    
1217
  @locking.ssynchronized(_config_lock, shared=1)
1218
  def GetInstanceInfo(self, instance_name):
1219
    """Returns information about an instance.
1220

1221
    It takes the information from the configuration file. Other information of
1222
    an instance are taken from the live systems.
1223

1224
    @param instance_name: name of the instance, e.g.
1225
        I{instance1.example.com}
1226

1227
    @rtype: L{objects.Instance}
1228
    @return: the instance object
1229

1230
    """
1231
    return self._UnlockedGetInstanceInfo(instance_name)
1232

    
1233
  @locking.ssynchronized(_config_lock, shared=1)
1234
  def GetAllInstancesInfo(self):
1235
    """Get the configuration of all instances.
1236

1237
    @rtype: dict
1238
    @return: dict of (instance, instance_info), where instance_info is what
1239
              would GetInstanceInfo return for the node
1240

1241
    """
1242
    my_dict = dict([(instance, self._UnlockedGetInstanceInfo(instance))
1243
                    for instance in self._UnlockedGetInstanceList()])
1244
    return my_dict
1245

    
1246
  @locking.ssynchronized(_config_lock)
1247
  def AddNode(self, node, ec_id):
1248
    """Add a node to the configuration.
1249

1250
    @type node: L{objects.Node}
1251
    @param node: a Node instance
1252

1253
    """
1254
    logging.info("Adding node %s to configuration", node.name)
1255

    
1256
    self._EnsureUUID(node, ec_id)
1257

    
1258
    node.serial_no = 1
1259
    node.ctime = node.mtime = time.time()
1260
    self._UnlockedAddNodeToGroup(node.name, node.group)
1261
    self._config_data.nodes[node.name] = node
1262
    self._config_data.cluster.serial_no += 1
1263
    self._WriteConfig()
1264

    
1265
  @locking.ssynchronized(_config_lock)
1266
  def RemoveNode(self, node_name):
1267
    """Remove a node from the configuration.
1268

1269
    """
1270
    logging.info("Removing node %s from configuration", node_name)
1271

    
1272
    if node_name not in self._config_data.nodes:
1273
      raise errors.ConfigurationError("Unknown node '%s'" % node_name)
1274

    
1275
    self._UnlockedRemoveNodeFromGroup(self._config_data.nodes[node_name])
1276
    del self._config_data.nodes[node_name]
1277
    self._config_data.cluster.serial_no += 1
1278
    self._WriteConfig()
1279

    
1280
  @locking.ssynchronized(_config_lock, shared=1)
1281
  def ExpandNodeName(self, short_name):
1282
    """Attempt to expand an incomplete instance name.
1283

1284
    """
1285
    return utils.MatchNameComponent(short_name,
1286
                                    self._config_data.nodes.keys(),
1287
                                    case_sensitive=False)
1288

    
1289
  def _UnlockedGetNodeInfo(self, node_name):
1290
    """Get the configuration of a node, as stored in the config.
1291

1292
    This function is for internal use, when the config lock is already
1293
    held.
1294

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

1297
    @rtype: L{objects.Node}
1298
    @return: the node object
1299

1300
    """
1301
    if node_name not in self._config_data.nodes:
1302
      return None
1303

    
1304
    return self._config_data.nodes[node_name]
1305

    
1306
  @locking.ssynchronized(_config_lock, shared=1)
1307
  def GetNodeInfo(self, node_name):
1308
    """Get the configuration of a node, as stored in the config.
1309

1310
    This is just a locked wrapper over L{_UnlockedGetNodeInfo}.
1311

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

1314
    @rtype: L{objects.Node}
1315
    @return: the node object
1316

1317
    """
1318
    return self._UnlockedGetNodeInfo(node_name)
1319

    
1320
  @locking.ssynchronized(_config_lock, shared=1)
1321
  def GetNodeInstances(self, node_name):
1322
    """Get the instances of a node, as stored in the config.
1323

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

1326
    @rtype: (list, list)
1327
    @return: a tuple with two lists: the primary and the secondary instances
1328

1329
    """
1330
    pri = []
1331
    sec = []
1332
    for inst in self._config_data.instances.values():
1333
      if inst.primary_node == node_name:
1334
        pri.append(inst.name)
1335
      if node_name in inst.secondary_nodes:
1336
        sec.append(inst.name)
1337
    return (pri, sec)
1338

    
1339
  def _UnlockedGetNodeList(self):
1340
    """Return the list of nodes which are in the configuration.
1341

1342
    This function is for internal use, when the config lock is already
1343
    held.
1344

1345
    @rtype: list
1346

1347
    """
1348
    return self._config_data.nodes.keys()
1349

    
1350
  @locking.ssynchronized(_config_lock, shared=1)
1351
  def GetNodeList(self):
1352
    """Return the list of nodes which are in the configuration.
1353

1354
    """
1355
    return self._UnlockedGetNodeList()
1356

    
1357
  def _UnlockedGetOnlineNodeList(self):
1358
    """Return the list of nodes which are online.
1359

1360
    """
1361
    all_nodes = [self._UnlockedGetNodeInfo(node)
1362
                 for node in self._UnlockedGetNodeList()]
1363
    return [node.name for node in all_nodes if not node.offline]
1364

    
1365
  @locking.ssynchronized(_config_lock, shared=1)
1366
  def GetOnlineNodeList(self):
1367
    """Return the list of nodes which are online.
1368

1369
    """
1370
    return self._UnlockedGetOnlineNodeList()
1371

    
1372
  @locking.ssynchronized(_config_lock, shared=1)
1373
  def GetVmCapableNodeList(self):
1374
    """Return the list of nodes which are not vm capable.
1375

1376
    """
1377
    all_nodes = [self._UnlockedGetNodeInfo(node)
1378
                 for node in self._UnlockedGetNodeList()]
1379
    return [node.name for node in all_nodes if node.vm_capable]
1380

    
1381
  @locking.ssynchronized(_config_lock, shared=1)
1382
  def GetNonVmCapableNodeList(self):
1383
    """Return the list of nodes which are not vm capable.
1384

1385
    """
1386
    all_nodes = [self._UnlockedGetNodeInfo(node)
1387
                 for node in self._UnlockedGetNodeList()]
1388
    return [node.name for node in all_nodes if not node.vm_capable]
1389

    
1390
  @locking.ssynchronized(_config_lock, shared=1)
1391
  def GetAllNodesInfo(self):
1392
    """Get the configuration of all nodes.
1393

1394
    @rtype: dict
1395
    @return: dict of (node, node_info), where node_info is what
1396
              would GetNodeInfo return for the node
1397

1398
    """
1399
    my_dict = dict([(node, self._UnlockedGetNodeInfo(node))
1400
                    for node in self._UnlockedGetNodeList()])
1401
    return my_dict
1402

    
1403
  @locking.ssynchronized(_config_lock, shared=1)
1404
  def GetNodeGroupsFromNodes(self, nodes):
1405
    """Returns groups for a list of nodes.
1406

1407
    @type nodes: list of string
1408
    @param nodes: List of node names
1409
    @rtype: frozenset
1410

1411
    """
1412
    return frozenset(self._UnlockedGetNodeInfo(name).group for name in nodes)
1413

    
1414
  def _UnlockedGetMasterCandidateStats(self, exceptions=None):
1415
    """Get the number of current and maximum desired and possible candidates.
1416

1417
    @type exceptions: list
1418
    @param exceptions: if passed, list of nodes that should be ignored
1419
    @rtype: tuple
1420
    @return: tuple of (current, desired and possible, possible)
1421

1422
    """
1423
    mc_now = mc_should = mc_max = 0
1424
    for node in self._config_data.nodes.values():
1425
      if exceptions and node.name in exceptions:
1426
        continue
1427
      if not (node.offline or node.drained) and node.master_capable:
1428
        mc_max += 1
1429
      if node.master_candidate:
1430
        mc_now += 1
1431
    mc_should = min(mc_max, self._config_data.cluster.candidate_pool_size)
1432
    return (mc_now, mc_should, mc_max)
1433

    
1434
  @locking.ssynchronized(_config_lock, shared=1)
1435
  def GetMasterCandidateStats(self, exceptions=None):
1436
    """Get the number of current and maximum possible candidates.
1437

1438
    This is just a wrapper over L{_UnlockedGetMasterCandidateStats}.
1439

1440
    @type exceptions: list
1441
    @param exceptions: if passed, list of nodes that should be ignored
1442
    @rtype: tuple
1443
    @return: tuple of (current, max)
1444

1445
    """
1446
    return self._UnlockedGetMasterCandidateStats(exceptions)
1447

    
1448
  @locking.ssynchronized(_config_lock)
1449
  def MaintainCandidatePool(self, exceptions):
1450
    """Try to grow the candidate pool to the desired size.
1451

1452
    @type exceptions: list
1453
    @param exceptions: if passed, list of nodes that should be ignored
1454
    @rtype: list
1455
    @return: list with the adjusted nodes (L{objects.Node} instances)
1456

1457
    """
1458
    mc_now, mc_max, _ = self._UnlockedGetMasterCandidateStats(exceptions)
1459
    mod_list = []
1460
    if mc_now < mc_max:
1461
      node_list = self._config_data.nodes.keys()
1462
      random.shuffle(node_list)
1463
      for name in node_list:
1464
        if mc_now >= mc_max:
1465
          break
1466
        node = self._config_data.nodes[name]
1467
        if (node.master_candidate or node.offline or node.drained or
1468
            node.name in exceptions or not node.master_capable):
1469
          continue
1470
        mod_list.append(node)
1471
        node.master_candidate = True
1472
        node.serial_no += 1
1473
        mc_now += 1
1474
      if mc_now != mc_max:
1475
        # this should not happen
1476
        logging.warning("Warning: MaintainCandidatePool didn't manage to"
1477
                        " fill the candidate pool (%d/%d)", mc_now, mc_max)
1478
      if mod_list:
1479
        self._config_data.cluster.serial_no += 1
1480
        self._WriteConfig()
1481

    
1482
    return mod_list
1483

    
1484
  def _UnlockedAddNodeToGroup(self, node_name, nodegroup_uuid):
1485
    """Add a given node to the specified group.
1486

1487
    """
1488
    if nodegroup_uuid not in self._config_data.nodegroups:
1489
      # This can happen if a node group gets deleted between its lookup and
1490
      # when we're adding the first node to it, since we don't keep a lock in
1491
      # the meantime. It's ok though, as we'll fail cleanly if the node group
1492
      # is not found anymore.
1493
      raise errors.OpExecError("Unknown node group: %s" % nodegroup_uuid)
1494
    if node_name not in self._config_data.nodegroups[nodegroup_uuid].members:
1495
      self._config_data.nodegroups[nodegroup_uuid].members.append(node_name)
1496

    
1497
  def _UnlockedRemoveNodeFromGroup(self, node):
1498
    """Remove a given node from its group.
1499

1500
    """
1501
    nodegroup = node.group
1502
    if nodegroup not in self._config_data.nodegroups:
1503
      logging.warning("Warning: node '%s' has unknown node group '%s'"
1504
                      " (while being removed from it)", node.name, nodegroup)
1505
    nodegroup_obj = self._config_data.nodegroups[nodegroup]
1506
    if node.name not in nodegroup_obj.members:
1507
      logging.warning("Warning: node '%s' not a member of its node group '%s'"
1508
                      " (while being removed from it)", node.name, nodegroup)
1509
    else:
1510
      nodegroup_obj.members.remove(node.name)
1511

    
1512
  def _BumpSerialNo(self):
1513
    """Bump up the serial number of the config.
1514

1515
    """
1516
    self._config_data.serial_no += 1
1517
    self._config_data.mtime = time.time()
1518

    
1519
  def _AllUUIDObjects(self):
1520
    """Returns all objects with uuid attributes.
1521

1522
    """
1523
    return (self._config_data.instances.values() +
1524
            self._config_data.nodes.values() +
1525
            self._config_data.nodegroups.values() +
1526
            [self._config_data.cluster])
1527

    
1528
  def _OpenConfig(self, accept_foreign):
1529
    """Read the config data from disk.
1530

1531
    """
1532
    raw_data = utils.ReadFile(self._cfg_file)
1533

    
1534
    try:
1535
      data = objects.ConfigData.FromDict(serializer.Load(raw_data))
1536
    except Exception, err:
1537
      raise errors.ConfigurationError(err)
1538

    
1539
    # Make sure the configuration has the right version
1540
    _ValidateConfig(data)
1541

    
1542
    if (not hasattr(data, 'cluster') or
1543
        not hasattr(data.cluster, 'rsahostkeypub')):
1544
      raise errors.ConfigurationError("Incomplete configuration"
1545
                                      " (missing cluster.rsahostkeypub)")
1546

    
1547
    if data.cluster.master_node != self._my_hostname and not accept_foreign:
1548
      msg = ("The configuration denotes node %s as master, while my"
1549
             " hostname is %s; opening a foreign configuration is only"
1550
             " possible in accept_foreign mode" %
1551
             (data.cluster.master_node, self._my_hostname))
1552
      raise errors.ConfigurationError(msg)
1553

    
1554
    # Upgrade configuration if needed
1555
    data.UpgradeConfig()
1556

    
1557
    self._config_data = data
1558
    # reset the last serial as -1 so that the next write will cause
1559
    # ssconf update
1560
    self._last_cluster_serial = -1
1561

    
1562
    # And finally run our (custom) config upgrade sequence
1563
    self._UpgradeConfig()
1564

    
1565
    self._cfg_id = utils.GetFileID(path=self._cfg_file)
1566

    
1567
  def _UpgradeConfig(self):
1568
    """Run upgrade steps that cannot be done purely in the objects.
1569

1570
    This is because some data elements need uniqueness across the
1571
    whole configuration, etc.
1572

1573
    @warning: this function will call L{_WriteConfig()}, but also
1574
        L{DropECReservations} so it needs to be called only from a
1575
        "safe" place (the constructor). If one wanted to call it with
1576
        the lock held, a DropECReservationUnlocked would need to be
1577
        created first, to avoid causing deadlock.
1578

1579
    """
1580
    modified = False
1581
    for item in self._AllUUIDObjects():
1582
      if item.uuid is None:
1583
        item.uuid = self._GenerateUniqueID(_UPGRADE_CONFIG_JID)
1584
        modified = True
1585
    if not self._config_data.nodegroups:
1586
      default_nodegroup_name = constants.INITIAL_NODE_GROUP_NAME
1587
      default_nodegroup = objects.NodeGroup(name=default_nodegroup_name,
1588
                                            members=[])
1589
      self._UnlockedAddNodeGroup(default_nodegroup, _UPGRADE_CONFIG_JID, True)
1590
      modified = True
1591
    for node in self._config_data.nodes.values():
1592
      if not node.group:
1593
        node.group = self.LookupNodeGroup(None)
1594
        modified = True
1595
      # This is technically *not* an upgrade, but needs to be done both when
1596
      # nodegroups are being added, and upon normally loading the config,
1597
      # because the members list of a node group is discarded upon
1598
      # serializing/deserializing the object.
1599
      self._UnlockedAddNodeToGroup(node.name, node.group)
1600
    if modified:
1601
      self._WriteConfig()
1602
      # This is ok even if it acquires the internal lock, as _UpgradeConfig is
1603
      # only called at config init time, without the lock held
1604
      self.DropECReservations(_UPGRADE_CONFIG_JID)
1605

    
1606
  def _DistributeConfig(self, feedback_fn):
1607
    """Distribute the configuration to the other nodes.
1608

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

1612
    """
1613
    if self._offline:
1614
      return True
1615

    
1616
    bad = False
1617

    
1618
    node_list = []
1619
    addr_list = []
1620
    myhostname = self._my_hostname
1621
    # we can skip checking whether _UnlockedGetNodeInfo returns None
1622
    # since the node list comes from _UnlocketGetNodeList, and we are
1623
    # called with the lock held, so no modifications should take place
1624
    # in between
1625
    for node_name in self._UnlockedGetNodeList():
1626
      if node_name == myhostname:
1627
        continue
1628
      node_info = self._UnlockedGetNodeInfo(node_name)
1629
      if not node_info.master_candidate:
1630
        continue
1631
      node_list.append(node_info.name)
1632
      addr_list.append(node_info.primary_ip)
1633

    
1634
    result = rpc.RpcRunner.call_upload_file(node_list, self._cfg_file,
1635
                                            address_list=addr_list)
1636
    for to_node, to_result in result.items():
1637
      msg = to_result.fail_msg
1638
      if msg:
1639
        msg = ("Copy of file %s to node %s failed: %s" %
1640
               (self._cfg_file, to_node, msg))
1641
        logging.error(msg)
1642

    
1643
        if feedback_fn:
1644
          feedback_fn(msg)
1645

    
1646
        bad = True
1647

    
1648
    return not bad
1649

    
1650
  def _WriteConfig(self, destination=None, feedback_fn=None):
1651
    """Write the configuration data to persistent storage.
1652

1653
    """
1654
    assert feedback_fn is None or callable(feedback_fn)
1655

    
1656
    # Warn on config errors, but don't abort the save - the
1657
    # configuration has already been modified, and we can't revert;
1658
    # the best we can do is to warn the user and save as is, leaving
1659
    # recovery to the user
1660
    config_errors = self._UnlockedVerifyConfig()
1661
    if config_errors:
1662
      errmsg = ("Configuration data is not consistent: %s" %
1663
                (utils.CommaJoin(config_errors)))
1664
      logging.critical(errmsg)
1665
      if feedback_fn:
1666
        feedback_fn(errmsg)
1667

    
1668
    if destination is None:
1669
      destination = self._cfg_file
1670
    self._BumpSerialNo()
1671
    txt = serializer.Dump(self._config_data.ToDict())
1672

    
1673
    getents = self._getents()
1674
    try:
1675
      fd = utils.SafeWriteFile(destination, self._cfg_id, data=txt,
1676
                               close=False, gid=getents.confd_gid, mode=0640)
1677
    except errors.LockError:
1678
      raise errors.ConfigurationError("The configuration file has been"
1679
                                      " modified since the last write, cannot"
1680
                                      " update")
1681
    try:
1682
      self._cfg_id = utils.GetFileID(fd=fd)
1683
    finally:
1684
      os.close(fd)
1685

    
1686
    self.write_count += 1
1687

    
1688
    # and redistribute the config file to master candidates
1689
    self._DistributeConfig(feedback_fn)
1690

    
1691
    # Write ssconf files on all nodes (including locally)
1692
    if self._last_cluster_serial < self._config_data.cluster.serial_no:
1693
      if not self._offline:
1694
        result = rpc.RpcRunner.call_write_ssconf_files(
1695
          self._UnlockedGetOnlineNodeList(),
1696
          self._UnlockedGetSsconfValues())
1697

    
1698
        for nname, nresu in result.items():
1699
          msg = nresu.fail_msg
1700
          if msg:
1701
            errmsg = ("Error while uploading ssconf files to"
1702
                      " node %s: %s" % (nname, msg))
1703
            logging.warning(errmsg)
1704

    
1705
            if feedback_fn:
1706
              feedback_fn(errmsg)
1707

    
1708
      self._last_cluster_serial = self._config_data.cluster.serial_no
1709

    
1710
  def _UnlockedGetSsconfValues(self):
1711
    """Return the values needed by ssconf.
1712

1713
    @rtype: dict
1714
    @return: a dictionary with keys the ssconf names and values their
1715
        associated value
1716

1717
    """
1718
    fn = "\n".join
1719
    instance_names = utils.NiceSort(self._UnlockedGetInstanceList())
1720
    node_names = utils.NiceSort(self._UnlockedGetNodeList())
1721
    node_info = [self._UnlockedGetNodeInfo(name) for name in node_names]
1722
    node_pri_ips = ["%s %s" % (ninfo.name, ninfo.primary_ip)
1723
                    for ninfo in node_info]
1724
    node_snd_ips = ["%s %s" % (ninfo.name, ninfo.secondary_ip)
1725
                    for ninfo in node_info]
1726

    
1727
    instance_data = fn(instance_names)
1728
    off_data = fn(node.name for node in node_info if node.offline)
1729
    on_data = fn(node.name for node in node_info if not node.offline)
1730
    mc_data = fn(node.name for node in node_info if node.master_candidate)
1731
    mc_ips_data = fn(node.primary_ip for node in node_info
1732
                     if node.master_candidate)
1733
    node_data = fn(node_names)
1734
    node_pri_ips_data = fn(node_pri_ips)
1735
    node_snd_ips_data = fn(node_snd_ips)
1736

    
1737
    cluster = self._config_data.cluster
1738
    cluster_tags = fn(cluster.GetTags())
1739

    
1740
    hypervisor_list = fn(cluster.enabled_hypervisors)
1741

    
1742
    uid_pool = uidpool.FormatUidPool(cluster.uid_pool, separator="\n")
1743

    
1744
    nodegroups = ["%s %s" % (nodegroup.uuid, nodegroup.name) for nodegroup in
1745
                  self._config_data.nodegroups.values()]
1746
    nodegroups_data = fn(utils.NiceSort(nodegroups))
1747

    
1748
    return {
1749
      constants.SS_CLUSTER_NAME: cluster.cluster_name,
1750
      constants.SS_CLUSTER_TAGS: cluster_tags,
1751
      constants.SS_FILE_STORAGE_DIR: cluster.file_storage_dir,
1752
      constants.SS_MASTER_CANDIDATES: mc_data,
1753
      constants.SS_MASTER_CANDIDATES_IPS: mc_ips_data,
1754
      constants.SS_MASTER_IP: cluster.master_ip,
1755
      constants.SS_MASTER_NETDEV: cluster.master_netdev,
1756
      constants.SS_MASTER_NODE: cluster.master_node,
1757
      constants.SS_NODE_LIST: node_data,
1758
      constants.SS_NODE_PRIMARY_IPS: node_pri_ips_data,
1759
      constants.SS_NODE_SECONDARY_IPS: node_snd_ips_data,
1760
      constants.SS_OFFLINE_NODES: off_data,
1761
      constants.SS_ONLINE_NODES: on_data,
1762
      constants.SS_PRIMARY_IP_FAMILY: str(cluster.primary_ip_family),
1763
      constants.SS_INSTANCE_LIST: instance_data,
1764
      constants.SS_RELEASE_VERSION: constants.RELEASE_VERSION,
1765
      constants.SS_HYPERVISOR_LIST: hypervisor_list,
1766
      constants.SS_MAINTAIN_NODE_HEALTH: str(cluster.maintain_node_health),
1767
      constants.SS_UID_POOL: uid_pool,
1768
      constants.SS_NODEGROUPS: nodegroups_data,
1769
      }
1770

    
1771
  @locking.ssynchronized(_config_lock, shared=1)
1772
  def GetSsconfValues(self):
1773
    """Wrapper using lock around _UnlockedGetSsconf().
1774

1775
    """
1776
    return self._UnlockedGetSsconfValues()
1777

    
1778
  @locking.ssynchronized(_config_lock, shared=1)
1779
  def GetVGName(self):
1780
    """Return the volume group name.
1781

1782
    """
1783
    return self._config_data.cluster.volume_group_name
1784

    
1785
  @locking.ssynchronized(_config_lock)
1786
  def SetVGName(self, vg_name):
1787
    """Set the volume group name.
1788

1789
    """
1790
    self._config_data.cluster.volume_group_name = vg_name
1791
    self._config_data.cluster.serial_no += 1
1792
    self._WriteConfig()
1793

    
1794
  @locking.ssynchronized(_config_lock, shared=1)
1795
  def GetDRBDHelper(self):
1796
    """Return DRBD usermode helper.
1797

1798
    """
1799
    return self._config_data.cluster.drbd_usermode_helper
1800

    
1801
  @locking.ssynchronized(_config_lock)
1802
  def SetDRBDHelper(self, drbd_helper):
1803
    """Set DRBD usermode helper.
1804

1805
    """
1806
    self._config_data.cluster.drbd_usermode_helper = drbd_helper
1807
    self._config_data.cluster.serial_no += 1
1808
    self._WriteConfig()
1809

    
1810
  @locking.ssynchronized(_config_lock, shared=1)
1811
  def GetMACPrefix(self):
1812
    """Return the mac prefix.
1813

1814
    """
1815
    return self._config_data.cluster.mac_prefix
1816

    
1817
  @locking.ssynchronized(_config_lock, shared=1)
1818
  def GetClusterInfo(self):
1819
    """Returns information about the cluster
1820

1821
    @rtype: L{objects.Cluster}
1822
    @return: the cluster object
1823

1824
    """
1825
    return self._config_data.cluster
1826

    
1827
  @locking.ssynchronized(_config_lock, shared=1)
1828
  def HasAnyDiskOfType(self, dev_type):
1829
    """Check if in there is at disk of the given type in the configuration.
1830

1831
    """
1832
    return self._config_data.HasAnyDiskOfType(dev_type)
1833

    
1834
  @locking.ssynchronized(_config_lock)
1835
  def Update(self, target, feedback_fn):
1836
    """Notify function to be called after updates.
1837

1838
    This function must be called when an object (as returned by
1839
    GetInstanceInfo, GetNodeInfo, GetCluster) has been updated and the
1840
    caller wants the modifications saved to the backing store. Note
1841
    that all modified objects will be saved, but the target argument
1842
    is the one the caller wants to ensure that it's saved.
1843

1844
    @param target: an instance of either L{objects.Cluster},
1845
        L{objects.Node} or L{objects.Instance} which is existing in
1846
        the cluster
1847
    @param feedback_fn: Callable feedback function
1848

1849
    """
1850
    if self._config_data is None:
1851
      raise errors.ProgrammerError("Configuration file not read,"
1852
                                   " cannot save.")
1853
    update_serial = False
1854
    if isinstance(target, objects.Cluster):
1855
      test = target == self._config_data.cluster
1856
    elif isinstance(target, objects.Node):
1857
      test = target in self._config_data.nodes.values()
1858
      update_serial = True
1859
    elif isinstance(target, objects.Instance):
1860
      test = target in self._config_data.instances.values()
1861
    elif isinstance(target, objects.NodeGroup):
1862
      test = target in self._config_data.nodegroups.values()
1863
    else:
1864
      raise errors.ProgrammerError("Invalid object type (%s) passed to"
1865
                                   " ConfigWriter.Update" % type(target))
1866
    if not test:
1867
      raise errors.ConfigurationError("Configuration updated since object"
1868
                                      " has been read or unknown object")
1869
    target.serial_no += 1
1870
    target.mtime = now = time.time()
1871

    
1872
    if update_serial:
1873
      # for node updates, we need to increase the cluster serial too
1874
      self._config_data.cluster.serial_no += 1
1875
      self._config_data.cluster.mtime = now
1876

    
1877
    if isinstance(target, objects.Instance):
1878
      self._UnlockedReleaseDRBDMinors(target.name)
1879

    
1880
    self._WriteConfig(feedback_fn=feedback_fn)
1881

    
1882
  @locking.ssynchronized(_config_lock)
1883
  def DropECReservations(self, ec_id):
1884
    """Drop per-execution-context reservations
1885

1886
    """
1887
    for rm in self._all_rms:
1888
      rm.DropECReservations(ec_id)