031c440f899350fd565f9f67e709c5a4e2a378cf
[ganeti-local] / lib / config.py
1 #
2 #
3
4 # Copyright (C) 2006, 2007 Google Inc.
5 #
6 # This program is free software; you can redistribute it and/or modify
7 # it under the terms of the GNU General Public License as published by
8 # the Free Software Foundation; either version 2 of the License, or
9 # (at your option) any later version.
10 #
11 # This program is distributed in the hope that it will be useful, but
12 # WITHOUT ANY WARRANTY; without even the implied warranty of
13 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
14 # General Public License for more details.
15 #
16 # You should have received a copy of the GNU General Public License
17 # along with this program; if not, write to the Free Software
18 # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
19 # 02110-1301, USA.
20
21
22 """Configuration management for Ganeti
23
24 This module provides the interface to the Ganeti cluster configuration.
25
26 The configuration data is stored on every node but is updated on the master
27 only. After each update, the master distributes the data to the other nodes.
28
29 Currently, the data storage format is JSON. YAML was slow and consuming too
30 much memory.
31
32 """
33
34 import os
35 import tempfile
36 import random
37 import logging
38
39 from ganeti import errors
40 from ganeti import locking
41 from ganeti import utils
42 from ganeti import constants
43 from ganeti import rpc
44 from ganeti import objects
45 from ganeti import serializer
46
47
48 _config_lock = locking.SharedLock()
49
50
51 def _ValidateConfig(data):
52   if data.version != constants.CONFIG_VERSION:
53     raise errors.ConfigurationError("Cluster configuration version"
54                                     " mismatch, got %s instead of %s" %
55                                     (data.version,
56                                      constants.CONFIG_VERSION))
57
58
59 class ConfigWriter:
60   """The interface to the cluster configuration.
61
62   """
63   def __init__(self, cfg_file=None, offline=False):
64     self.write_count = 0
65     self._lock = _config_lock
66     self._config_data = None
67     self._offline = offline
68     if cfg_file is None:
69       self._cfg_file = constants.CLUSTER_CONF_FILE
70     else:
71       self._cfg_file = cfg_file
72     self._temporary_ids = set()
73     self._temporary_drbds = {}
74     # Note: in order to prevent errors when resolving our name in
75     # _DistributeConfig, we compute it here once and reuse it; it's
76     # better to raise an error before starting to modify the config
77     # file than after it was modified
78     self._my_hostname = utils.HostInfo().name
79     self._OpenConfig()
80
81   # this method needs to be static, so that we can call it on the class
82   @staticmethod
83   def IsCluster():
84     """Check if the cluster is configured.
85
86     """
87     return os.path.exists(constants.CLUSTER_CONF_FILE)
88
89   @locking.ssynchronized(_config_lock, shared=1)
90   def GenerateMAC(self):
91     """Generate a MAC for an instance.
92
93     This should check the current instances for duplicates.
94
95     """
96     prefix = self._config_data.cluster.mac_prefix
97     all_macs = self._AllMACs()
98     retries = 64
99     while retries > 0:
100       byte1 = random.randrange(0, 256)
101       byte2 = random.randrange(0, 256)
102       byte3 = random.randrange(0, 256)
103       mac = "%s:%02x:%02x:%02x" % (prefix, byte1, byte2, byte3)
104       if mac not in all_macs:
105         break
106       retries -= 1
107     else:
108       raise errors.ConfigurationError("Can't generate unique MAC")
109     return mac
110
111   @locking.ssynchronized(_config_lock, shared=1)
112   def IsMacInUse(self, mac):
113     """Predicate: check if the specified MAC is in use in the Ganeti cluster.
114
115     This only checks instances managed by this cluster, it does not
116     check for potential collisions elsewhere.
117
118     """
119     all_macs = self._AllMACs()
120     return mac in all_macs
121
122   @locking.ssynchronized(_config_lock, shared=1)
123   def GenerateDRBDSecret(self):
124     """Generate a DRBD secret.
125
126     This checks the current disks for duplicates.
127
128     """
129     all_secrets = self._AllDRBDSecrets()
130     retries = 64
131     while retries > 0:
132       secret = utils.GenerateSecret()
133       if secret not in all_secrets:
134         break
135       retries -= 1
136     else:
137       raise errors.ConfigurationError("Can't generate unique DRBD secret")
138     return secret
139
140   def _ComputeAllLVs(self):
141     """Compute the list of all LVs.
142
143     """
144     lvnames = set()
145     for instance in self._config_data.instances.values():
146       node_data = instance.MapLVsByNode()
147       for lv_list in node_data.values():
148         lvnames.update(lv_list)
149     return lvnames
150
151   @locking.ssynchronized(_config_lock, shared=1)
152   def GenerateUniqueID(self, exceptions=None):
153     """Generate an unique disk name.
154
155     This checks the current node, instances and disk names for
156     duplicates.
157
158     Args:
159       - exceptions: a list with some other names which should be checked
160                     for uniqueness (used for example when you want to get
161                     more than one id at one time without adding each one in
162                     turn to the config file
163
164     Returns: the unique id as a string
165
166     """
167     existing = set()
168     existing.update(self._temporary_ids)
169     existing.update(self._ComputeAllLVs())
170     existing.update(self._config_data.instances.keys())
171     existing.update(self._config_data.nodes.keys())
172     if exceptions is not None:
173       existing.update(exceptions)
174     retries = 64
175     while retries > 0:
176       unique_id = utils.NewUUID()
177       if unique_id not in existing and unique_id is not None:
178         break
179     else:
180       raise errors.ConfigurationError("Not able generate an unique ID"
181                                       " (last tried ID: %s" % unique_id)
182     self._temporary_ids.add(unique_id)
183     return unique_id
184
185   def _AllMACs(self):
186     """Return all MACs present in the config.
187
188     """
189     result = []
190     for instance in self._config_data.instances.values():
191       for nic in instance.nics:
192         result.append(nic.mac)
193
194     return result
195
196   def _AllDRBDSecrets(self):
197     """Return all DRBD secrets present in the config.
198
199     """
200     def helper(disk, result):
201       """Recursively gather secrets from this disk."""
202       if disk.dev_type == constants.DT_DRBD8:
203         result.append(disk.logical_id[5])
204       if disk.children:
205         for child in disk.children:
206           helper(child, result)
207
208     result = []
209     for instance in self._config_data.instances.values():
210       for disk in instance.disks:
211         helper(disk, result)
212
213     return result
214
215   @locking.ssynchronized(_config_lock, shared=1)
216   def VerifyConfig(self):
217     """Stub verify function.
218     """
219     result = []
220     seen_macs = []
221     ports = {}
222     data = self._config_data
223     for instance_name in data.instances:
224       instance = data.instances[instance_name]
225       if instance.primary_node not in data.nodes:
226         result.append("instance '%s' has invalid primary node '%s'" %
227                       (instance_name, instance.primary_node))
228       for snode in instance.secondary_nodes:
229         if snode not in data.nodes:
230           result.append("instance '%s' has invalid secondary node '%s'" %
231                         (instance_name, snode))
232       for idx, nic in enumerate(instance.nics):
233         if nic.mac in seen_macs:
234           result.append("instance '%s' has NIC %d mac %s duplicate" %
235                         (instance_name, idx, nic.mac))
236         else:
237           seen_macs.append(nic.mac)
238
239       # gather the drbd ports for duplicate checks
240       for dsk in instance.disks:
241         if dsk.dev_type in constants.LDS_DRBD:
242           tcp_port = dsk.logical_id[2]
243           if tcp_port not in ports:
244             ports[tcp_port] = []
245           ports[tcp_port].append((instance.name, "drbd disk %s" % dsk.iv_name))
246       # gather network port reservation
247       net_port = getattr(instance, "network_port", None)
248       if net_port is not None:
249         if net_port not in ports:
250           ports[net_port] = []
251         ports[net_port].append((instance.name, "network port"))
252
253     # cluster-wide pool of free ports
254     for free_port in self._config_data.cluster.tcpudp_port_pool:
255       if free_port not in ports:
256         ports[free_port] = []
257       ports[free_port].append(("cluster", "port marked as free"))
258
259     # compute tcp/udp duplicate ports
260     keys = ports.keys()
261     keys.sort()
262     for pnum in keys:
263       pdata = ports[pnum]
264       if len(pdata) > 1:
265         txt = ", ".join(["%s/%s" % val for val in pdata])
266         result.append("tcp/udp port %s has duplicates: %s" % (pnum, txt))
267
268     # highest used tcp port check
269     if keys:
270       if keys[-1] > self._config_data.cluster.highest_used_port:
271         result.append("Highest used port mismatch, saved %s, computed %s" %
272                       (self._config_data.cluster.highest_used_port,
273                        keys[-1]))
274
275     return result
276
277   def _UnlockedSetDiskID(self, disk, node_name):
278     """Convert the unique ID to the ID needed on the target nodes.
279
280     This is used only for drbd, which needs ip/port configuration.
281
282     The routine descends down and updates its children also, because
283     this helps when the only the top device is passed to the remote
284     node.
285
286     This function is for internal use, when the config lock is already held.
287
288     """
289     if disk.children:
290       for child in disk.children:
291         self._UnlockedSetDiskID(child, node_name)
292
293     if disk.logical_id is None and disk.physical_id is not None:
294       return
295     if disk.dev_type == constants.LD_DRBD8:
296       pnode, snode, port, pminor, sminor, secret = disk.logical_id
297       if node_name not in (pnode, snode):
298         raise errors.ConfigurationError("DRBD device not knowing node %s" %
299                                         node_name)
300       pnode_info = self._UnlockedGetNodeInfo(pnode)
301       snode_info = self._UnlockedGetNodeInfo(snode)
302       if pnode_info is None or snode_info is None:
303         raise errors.ConfigurationError("Can't find primary or secondary node"
304                                         " for %s" % str(disk))
305       p_data = (pnode_info.secondary_ip, port)
306       s_data = (snode_info.secondary_ip, port)
307       if pnode == node_name:
308         disk.physical_id = p_data + s_data + (pminor, secret)
309       else: # it must be secondary, we tested above
310         disk.physical_id = s_data + p_data + (sminor, secret)
311     else:
312       disk.physical_id = disk.logical_id
313     return
314
315   @locking.ssynchronized(_config_lock)
316   def SetDiskID(self, disk, node_name):
317     """Convert the unique ID to the ID needed on the target nodes.
318
319     This is used only for drbd, which needs ip/port configuration.
320
321     The routine descends down and updates its children also, because
322     this helps when the only the top device is passed to the remote
323     node.
324
325     """
326     return self._UnlockedSetDiskID(disk, node_name)
327
328   @locking.ssynchronized(_config_lock)
329   def AddTcpUdpPort(self, port):
330     """Adds a new port to the available port pool.
331
332     """
333     if not isinstance(port, int):
334       raise errors.ProgrammerError("Invalid type passed for port")
335
336     self._config_data.cluster.tcpudp_port_pool.add(port)
337     self._WriteConfig()
338
339   @locking.ssynchronized(_config_lock, shared=1)
340   def GetPortList(self):
341     """Returns a copy of the current port list.
342
343     """
344     return self._config_data.cluster.tcpudp_port_pool.copy()
345
346   @locking.ssynchronized(_config_lock)
347   def AllocatePort(self):
348     """Allocate a port.
349
350     The port will be taken from the available port pool or from the
351     default port range (and in this case we increase
352     highest_used_port).
353
354     """
355     # If there are TCP/IP ports configured, we use them first.
356     if self._config_data.cluster.tcpudp_port_pool:
357       port = self._config_data.cluster.tcpudp_port_pool.pop()
358     else:
359       port = self._config_data.cluster.highest_used_port + 1
360       if port >= constants.LAST_DRBD_PORT:
361         raise errors.ConfigurationError("The highest used port is greater"
362                                         " than %s. Aborting." %
363                                         constants.LAST_DRBD_PORT)
364       self._config_data.cluster.highest_used_port = port
365
366     self._WriteConfig()
367     return port
368
369   def _ComputeDRBDMap(self, instance):
370     """Compute the used DRBD minor/nodes.
371
372     Return: dictionary of node_name: dict of minor: instance_name. The
373     returned dict will have all the nodes in it (even if with an empty
374     list).
375
376     """
377     def _AppendUsedPorts(instance_name, disk, used):
378       if disk.dev_type == constants.LD_DRBD8 and len(disk.logical_id) >= 5:
379         nodeA, nodeB, dummy, minorA, minorB = disk.logical_id[:5]
380         for node, port in ((nodeA, minorA), (nodeB, minorB)):
381           assert node in used, "Instance node not found in node list"
382           if port in used[node]:
383             raise errors.ProgrammerError("DRBD minor already used:"
384                                          " %s/%s, %s/%s" %
385                                          (node, port, instance_name,
386                                           used[node][port]))
387
388           used[node][port] = instance_name
389       if disk.children:
390         for child in disk.children:
391           _AppendUsedPorts(instance_name, child, used)
392
393     my_dict = dict((node, {}) for node in self._config_data.nodes)
394     for (node, minor), instance in self._temporary_drbds.iteritems():
395       my_dict[node][minor] = instance
396     for instance in self._config_data.instances.itervalues():
397       for disk in instance.disks:
398         _AppendUsedPorts(instance.name, disk, my_dict)
399     return my_dict
400
401   @locking.ssynchronized(_config_lock)
402   def AllocateDRBDMinor(self, nodes, instance):
403     """Allocate a drbd minor.
404
405     The free minor will be automatically computed from the existing
406     devices. A node can be given multiple times in order to allocate
407     multiple minors. The result is the list of minors, in the same
408     order as the passed nodes.
409
410     """
411     d_map = self._ComputeDRBDMap(instance)
412     result = []
413     for nname in nodes:
414       ndata = d_map[nname]
415       if not ndata:
416         # no minors used, we can start at 0
417         result.append(0)
418         ndata[0] = instance
419         self._temporary_drbds[(nname, 0)] = instance
420         continue
421       keys = ndata.keys()
422       keys.sort()
423       ffree = utils.FirstFree(keys)
424       if ffree is None:
425         # return the next minor
426         # TODO: implement high-limit check
427         minor = keys[-1] + 1
428       else:
429         minor = ffree
430       result.append(minor)
431       ndata[minor] = instance
432       assert (nname, minor) not in self._temporary_drbds, \
433              "Attempt to reuse reserved DRBD minor"
434       self._temporary_drbds[(nname, minor)] = instance
435     logging.debug("Request to allocate drbd minors, input: %s, returning %s",
436                   nodes, result)
437     return result
438
439   @locking.ssynchronized(_config_lock)
440   def ReleaseDRBDMinors(self, instance):
441     """Release temporary drbd minors allocated for a given instance.
442
443     This should be called on both the error paths and on the success
444     paths (after the instance has been added or updated).
445
446     @type instance: string
447     @param instance: the instance for which temporary minors should be
448                      released
449
450     """
451     for key, name in self._temporary_drbds.items():
452       if name == instance:
453         del self._temporary_drbds[key]
454
455   @locking.ssynchronized(_config_lock, shared=1)
456   def GetConfigVersion(self):
457     """Get the configuration version.
458
459     @return: Config version
460
461     """
462     return self._config_data.version
463
464   @locking.ssynchronized(_config_lock, shared=1)
465   def GetClusterName(self):
466     """Get cluster name.
467
468     @return: Cluster name
469
470     """
471     return self._config_data.cluster.cluster_name
472
473   @locking.ssynchronized(_config_lock, shared=1)
474   def GetMasterNode(self):
475     """Get the hostname of the master node for this cluster.
476
477     @return: Master hostname
478
479     """
480     return self._config_data.cluster.master_node
481
482   @locking.ssynchronized(_config_lock, shared=1)
483   def GetMasterIP(self):
484     """Get the IP of the master node for this cluster.
485
486     @return: Master IP
487
488     """
489     return self._config_data.cluster.master_ip
490
491   @locking.ssynchronized(_config_lock, shared=1)
492   def GetMasterNetdev(self):
493     """Get the master network device for this cluster.
494
495     """
496     return self._config_data.cluster.master_netdev
497
498   @locking.ssynchronized(_config_lock, shared=1)
499   def GetFileStorageDir(self):
500     """Get the file storage dir for this cluster.
501
502     """
503     return self._config_data.cluster.file_storage_dir
504
505   @locking.ssynchronized(_config_lock, shared=1)
506   def GetHypervisorType(self):
507     """Get the hypervisor type for this cluster.
508
509     """
510     return self._config_data.cluster.default_hypervisor
511
512   @locking.ssynchronized(_config_lock, shared=1)
513   def GetHostKey(self):
514     """Return the rsa hostkey from the config.
515
516     Args: None
517
518     Returns: rsa hostkey
519     """
520     return self._config_data.cluster.rsahostkeypub
521
522   @locking.ssynchronized(_config_lock)
523   def AddInstance(self, instance):
524     """Add an instance to the config.
525
526     This should be used after creating a new instance.
527
528     Args:
529       instance: the instance object
530     """
531     if not isinstance(instance, objects.Instance):
532       raise errors.ProgrammerError("Invalid type passed to AddInstance")
533
534     if instance.disk_template != constants.DT_DISKLESS:
535       all_lvs = instance.MapLVsByNode()
536       logging.info("Instance '%s' DISK_LAYOUT: %s", instance.name, all_lvs)
537
538     instance.serial_no = 1
539     self._config_data.instances[instance.name] = instance
540     self._config_data.cluster.serial_no += 1
541     self._WriteConfig()
542
543   def _SetInstanceStatus(self, instance_name, status):
544     """Set the instance's status to a given value.
545
546     """
547     if status not in ("up", "down"):
548       raise errors.ProgrammerError("Invalid status '%s' passed to"
549                                    " ConfigWriter._SetInstanceStatus()" %
550                                    status)
551
552     if instance_name not in self._config_data.instances:
553       raise errors.ConfigurationError("Unknown instance '%s'" %
554                                       instance_name)
555     instance = self._config_data.instances[instance_name]
556     if instance.status != status:
557       instance.status = status
558       instance.serial_no += 1
559       self._WriteConfig()
560
561   @locking.ssynchronized(_config_lock)
562   def MarkInstanceUp(self, instance_name):
563     """Mark the instance status to up in the config.
564
565     """
566     self._SetInstanceStatus(instance_name, "up")
567
568   @locking.ssynchronized(_config_lock)
569   def RemoveInstance(self, instance_name):
570     """Remove the instance from the configuration.
571
572     """
573     if instance_name not in self._config_data.instances:
574       raise errors.ConfigurationError("Unknown instance '%s'" % instance_name)
575     del self._config_data.instances[instance_name]
576     self._config_data.cluster.serial_no += 1
577     self._WriteConfig()
578
579   @locking.ssynchronized(_config_lock)
580   def RenameInstance(self, old_name, new_name):
581     """Rename an instance.
582
583     This needs to be done in ConfigWriter and not by RemoveInstance
584     combined with AddInstance as only we can guarantee an atomic
585     rename.
586
587     """
588     if old_name not in self._config_data.instances:
589       raise errors.ConfigurationError("Unknown instance '%s'" % old_name)
590     inst = self._config_data.instances[old_name]
591     del self._config_data.instances[old_name]
592     inst.name = new_name
593
594     for disk in inst.disks:
595       if disk.dev_type == constants.LD_FILE:
596         # rename the file paths in logical and physical id
597         file_storage_dir = os.path.dirname(os.path.dirname(disk.logical_id[1]))
598         disk.physical_id = disk.logical_id = (disk.logical_id[0],
599                                               os.path.join(file_storage_dir,
600                                                            inst.name,
601                                                            disk.iv_name))
602
603     self._config_data.instances[inst.name] = inst
604     self._config_data.cluster.serial_no += 1
605     self._WriteConfig()
606
607   @locking.ssynchronized(_config_lock)
608   def MarkInstanceDown(self, instance_name):
609     """Mark the status of an instance to down in the configuration.
610
611     """
612     self._SetInstanceStatus(instance_name, "down")
613
614   def _UnlockedGetInstanceList(self):
615     """Get the list of instances.
616
617     This function is for internal use, when the config lock is already held.
618
619     """
620     return self._config_data.instances.keys()
621
622   @locking.ssynchronized(_config_lock, shared=1)
623   def GetInstanceList(self):
624     """Get the list of instances.
625
626     Returns:
627       array of instances, ex. ['instance2.example.com','instance1.example.com']
628       these contains all the instances, also the ones in Admin_down state
629
630     """
631     return self._UnlockedGetInstanceList()
632
633   @locking.ssynchronized(_config_lock, shared=1)
634   def ExpandInstanceName(self, short_name):
635     """Attempt to expand an incomplete instance name.
636
637     """
638     return utils.MatchNameComponent(short_name,
639                                     self._config_data.instances.keys())
640
641   def _UnlockedGetInstanceInfo(self, instance_name):
642     """Returns informations about an instance.
643
644     This function is for internal use, when the config lock is already held.
645
646     """
647     if instance_name not in self._config_data.instances:
648       return None
649
650     return self._config_data.instances[instance_name]
651
652   @locking.ssynchronized(_config_lock, shared=1)
653   def GetInstanceInfo(self, instance_name):
654     """Returns informations about an instance.
655
656     It takes the information from the configuration file. Other informations of
657     an instance are taken from the live systems.
658
659     Args:
660       instance: name of the instance, ex instance1.example.com
661
662     Returns:
663       the instance object
664
665     """
666     return self._UnlockedGetInstanceInfo(instance_name)
667
668   @locking.ssynchronized(_config_lock, shared=1)
669   def GetAllInstancesInfo(self):
670     """Get the configuration of all instances.
671
672     @rtype: dict
673     @returns: dict of (instance, instance_info), where instance_info is what
674               would GetInstanceInfo return for the node
675
676     """
677     my_dict = dict([(instance, self._UnlockedGetInstanceInfo(instance))
678                     for instance in self._UnlockedGetInstanceList()])
679     return my_dict
680
681   @locking.ssynchronized(_config_lock)
682   def AddNode(self, node):
683     """Add a node to the configuration.
684
685     Args:
686       node: an object.Node instance
687
688     """
689     logging.info("Adding node %s to configuration" % node.name)
690
691     node.serial_no = 1
692     self._config_data.nodes[node.name] = node
693     self._config_data.cluster.serial_no += 1
694     self._WriteConfig()
695
696   @locking.ssynchronized(_config_lock)
697   def RemoveNode(self, node_name):
698     """Remove a node from the configuration.
699
700     """
701     logging.info("Removing node %s from configuration" % node_name)
702
703     if node_name not in self._config_data.nodes:
704       raise errors.ConfigurationError("Unknown node '%s'" % node_name)
705
706     del self._config_data.nodes[node_name]
707     self._config_data.cluster.serial_no += 1
708     self._WriteConfig()
709
710   @locking.ssynchronized(_config_lock, shared=1)
711   def ExpandNodeName(self, short_name):
712     """Attempt to expand an incomplete instance name.
713
714     """
715     return utils.MatchNameComponent(short_name,
716                                     self._config_data.nodes.keys())
717
718   def _UnlockedGetNodeInfo(self, node_name):
719     """Get the configuration of a node, as stored in the config.
720
721     This function is for internal use, when the config lock is already held.
722
723     Args: node: nodename (tuple) of the node
724
725     Returns: the node object
726
727     """
728     if node_name not in self._config_data.nodes:
729       return None
730
731     return self._config_data.nodes[node_name]
732
733
734   @locking.ssynchronized(_config_lock, shared=1)
735   def GetNodeInfo(self, node_name):
736     """Get the configuration of a node, as stored in the config.
737
738     Args: node: nodename (tuple) of the node
739
740     Returns: the node object
741
742     """
743     return self._UnlockedGetNodeInfo(node_name)
744
745   def _UnlockedGetNodeList(self):
746     """Return the list of nodes which are in the configuration.
747
748     This function is for internal use, when the config lock is already held.
749
750     """
751     return self._config_data.nodes.keys()
752
753
754   @locking.ssynchronized(_config_lock, shared=1)
755   def GetNodeList(self):
756     """Return the list of nodes which are in the configuration.
757
758     """
759     return self._UnlockedGetNodeList()
760
761   @locking.ssynchronized(_config_lock, shared=1)
762   def GetAllNodesInfo(self):
763     """Get the configuration of all nodes.
764
765     @rtype: dict
766     @returns: dict of (node, node_info), where node_info is what
767               would GetNodeInfo return for the node
768
769     """
770     my_dict = dict([(node, self._UnlockedGetNodeInfo(node))
771                     for node in self._UnlockedGetNodeList()])
772     return my_dict
773
774   def _BumpSerialNo(self):
775     """Bump up the serial number of the config.
776
777     """
778     self._config_data.serial_no += 1
779
780   def _OpenConfig(self):
781     """Read the config data from disk.
782
783     In case we already have configuration data and the config file has
784     the same mtime as when we read it, we skip the parsing of the
785     file, since de-serialisation could be slow.
786
787     """
788     f = open(self._cfg_file, 'r')
789     try:
790       try:
791         data = objects.ConfigData.FromDict(serializer.Load(f.read()))
792       except Exception, err:
793         raise errors.ConfigurationError(err)
794     finally:
795       f.close()
796
797     # Make sure the configuration has the right version
798     _ValidateConfig(data)
799
800     if (not hasattr(data, 'cluster') or
801         not hasattr(data.cluster, 'rsahostkeypub')):
802       raise errors.ConfigurationError("Incomplete configuration"
803                                       " (missing cluster.rsahostkeypub)")
804     self._config_data = data
805     # init the last serial as -1 so that the next write will cause
806     # ssconf update
807     self._last_cluster_serial = -1
808
809   def _DistributeConfig(self):
810     """Distribute the configuration to the other nodes.
811
812     Currently, this only copies the configuration file. In the future,
813     it could be used to encapsulate the 2/3-phase update mechanism.
814
815     """
816     if self._offline:
817       return True
818     bad = False
819     nodelist = self._UnlockedGetNodeList()
820     myhostname = self._my_hostname
821
822     try:
823       nodelist.remove(myhostname)
824     except ValueError:
825       pass
826     # we can skip checking whether _UnlockedGetNodeInfo returns None
827     # since the node list comes from _UnlocketGetNodeList, and we are
828     # called with the lock held, so no modifications should take place
829     # in between
830     address_list = [self._UnlockedGetNodeInfo(name).primary_ip
831                     for name in nodelist]
832
833     result = rpc.RpcRunner.call_upload_file(nodelist, self._cfg_file,
834                                             address_list=address_list)
835     for node in nodelist:
836       if not result[node]:
837         logging.error("copy of file %s to node %s failed",
838                       self._cfg_file, node)
839         bad = True
840     return not bad
841
842   def _WriteConfig(self, destination=None):
843     """Write the configuration data to persistent storage.
844
845     """
846     if destination is None:
847       destination = self._cfg_file
848     self._BumpSerialNo()
849     txt = serializer.Dump(self._config_data.ToDict())
850     dir_name, file_name = os.path.split(destination)
851     fd, name = tempfile.mkstemp('.newconfig', file_name, dir_name)
852     f = os.fdopen(fd, 'w')
853     try:
854       f.write(txt)
855       os.fsync(f.fileno())
856     finally:
857       f.close()
858     # we don't need to do os.close(fd) as f.close() did it
859     os.rename(name, destination)
860     self.write_count += 1
861
862     # and redistribute the config file
863     self._DistributeConfig()
864
865     # Write ssconf files on all nodes (including locally)
866     if self._last_cluster_serial < self._config_data.cluster.serial_no:
867       if not self._offline:
868         rpc.RpcRunner.call_write_ssconf_files(self._UnlockedGetNodeList(),
869                                               self._UnlockedGetSsconfValues())
870       self._last_cluster_serial = self._config_data.cluster.serial_no
871
872   def _UnlockedGetSsconfValues(self):
873     return {
874       "cluster_name": self._config_data.cluster.cluster_name,
875       "master_ip": self._config_data.cluster.master_ip,
876       "master_netdev": self._config_data.cluster.master_netdev,
877       "master_node": self._config_data.cluster.master_node,
878       }
879
880   @locking.ssynchronized(_config_lock)
881   def InitConfig(self, version, cluster_config, master_node_config):
882     """Create the initial cluster configuration.
883
884     It will contain the current node, which will also be the master
885     node, and no instances.
886
887     @type version: int
888     @param version: Configuration version
889     @type cluster_config: objects.Cluster
890     @param cluster_config: Cluster configuration
891     @type master_node_config: objects.Node
892     @param master_node_config: Master node configuration
893
894     """
895     nodes = {
896       master_node_config.name: master_node_config,
897       }
898
899     self._config_data = objects.ConfigData(version=version,
900                                            cluster=cluster_config,
901                                            nodes=nodes,
902                                            instances={},
903                                            serial_no=1)
904     self._WriteConfig()
905
906   @locking.ssynchronized(_config_lock, shared=1)
907   def GetVGName(self):
908     """Return the volume group name.
909
910     """
911     return self._config_data.cluster.volume_group_name
912
913   @locking.ssynchronized(_config_lock)
914   def SetVGName(self, vg_name):
915     """Set the volume group name.
916
917     """
918     self._config_data.cluster.volume_group_name = vg_name
919     self._config_data.cluster.serial_no += 1
920     self._WriteConfig()
921
922   @locking.ssynchronized(_config_lock, shared=1)
923   def GetDefBridge(self):
924     """Return the default bridge.
925
926     """
927     return self._config_data.cluster.default_bridge
928
929   @locking.ssynchronized(_config_lock, shared=1)
930   def GetMACPrefix(self):
931     """Return the mac prefix.
932
933     """
934     return self._config_data.cluster.mac_prefix
935
936   @locking.ssynchronized(_config_lock, shared=1)
937   def GetClusterInfo(self):
938     """Returns informations about the cluster
939
940     Returns:
941       the cluster object
942
943     """
944     return self._config_data.cluster
945
946   @locking.ssynchronized(_config_lock)
947   def Update(self, target):
948     """Notify function to be called after updates.
949
950     This function must be called when an object (as returned by
951     GetInstanceInfo, GetNodeInfo, GetCluster) has been updated and the
952     caller wants the modifications saved to the backing store. Note
953     that all modified objects will be saved, but the target argument
954     is the one the caller wants to ensure that it's saved.
955
956     """
957     if self._config_data is None:
958       raise errors.ProgrammerError("Configuration file not read,"
959                                    " cannot save.")
960     if isinstance(target, objects.Cluster):
961       test = target == self._config_data.cluster
962     elif isinstance(target, objects.Node):
963       test = target in self._config_data.nodes.values()
964     elif isinstance(target, objects.Instance):
965       test = target in self._config_data.instances.values()
966     else:
967       raise errors.ProgrammerError("Invalid object type (%s) passed to"
968                                    " ConfigWriter.Update" % type(target))
969     if not test:
970       raise errors.ConfigurationError("Configuration updated since object"
971                                       " has been read or unknown object")
972     target.serial_no += 1
973
974     self._WriteConfig()