4 # Copyright (C) 2006, 2007, 2008, 2010, 2011, 2012 Google Inc.
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.
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.
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
22 """Functions to bootstrap a new cluster.
33 from ganeti.cmdlib import cluster
34 from ganeti import rpc
35 from ganeti import ssh
36 from ganeti import utils
37 from ganeti import errors
38 from ganeti import config
39 from ganeti import constants
40 from ganeti import objects
41 from ganeti import ssconf
42 from ganeti import serializer
43 from ganeti import hypervisor
44 from ganeti.storage import drbd
45 from ganeti.storage import filestorage
46 from ganeti import netutils
47 from ganeti import luxi
48 from ganeti import jstore
49 from ganeti import pathutils
52 # ec_id for InitConfig's temporary reservation manager
53 _INITCONF_ECID = "initconfig-ecid"
55 #: After how many seconds daemon must be responsive
56 _DAEMON_READY_TIMEOUT = 10.0
60 """Setup the SSH configuration for the cluster.
62 This generates a dsa keypair for root, adds the pub key to the
63 permitted hosts and adds the hostkey to its own known hosts.
66 priv_key, pub_key, auth_keys = ssh.GetUserFiles(constants.SSH_LOGIN_USER)
68 for name in priv_key, pub_key:
69 if os.path.exists(name):
70 utils.CreateBackup(name)
71 utils.RemoveFile(name)
73 result = utils.RunCmd(["ssh-keygen", "-t", "dsa",
77 raise errors.OpExecError("Could not generate ssh keypair, error %s" %
80 utils.AddAuthorizedKey(auth_keys, utils.ReadFile(pub_key))
83 def GenerateHmacKey(file_name):
84 """Writes a new HMAC key.
87 @param file_name: Path to output file
90 utils.WriteFile(file_name, data="%s\n" % utils.GenerateSecret(), mode=0400,
94 def GenerateClusterCrypto(new_cluster_cert, new_rapi_cert, new_spice_cert,
95 new_confd_hmac_key, new_cds,
96 rapi_cert_pem=None, spice_cert_pem=None,
97 spice_cacert_pem=None, cds=None,
98 nodecert_file=pathutils.NODED_CERT_FILE,
99 rapicert_file=pathutils.RAPI_CERT_FILE,
100 spicecert_file=pathutils.SPICE_CERT_FILE,
101 spicecacert_file=pathutils.SPICE_CACERT_FILE,
102 hmackey_file=pathutils.CONFD_HMAC_KEY,
103 cds_file=pathutils.CLUSTER_DOMAIN_SECRET_FILE):
104 """Updates the cluster certificates, keys and secrets.
106 @type new_cluster_cert: bool
107 @param new_cluster_cert: Whether to generate a new cluster certificate
108 @type new_rapi_cert: bool
109 @param new_rapi_cert: Whether to generate a new RAPI certificate
110 @type new_spice_cert: bool
111 @param new_spice_cert: Whether to generate a new SPICE certificate
112 @type new_confd_hmac_key: bool
113 @param new_confd_hmac_key: Whether to generate a new HMAC key
115 @param new_cds: Whether to generate a new cluster domain secret
116 @type rapi_cert_pem: string
117 @param rapi_cert_pem: New RAPI certificate in PEM format
118 @type spice_cert_pem: string
119 @param spice_cert_pem: New SPICE certificate in PEM format
120 @type spice_cacert_pem: string
121 @param spice_cacert_pem: Certificate of the CA that signed the SPICE
122 certificate, in PEM format
124 @param cds: New cluster domain secret
125 @type nodecert_file: string
126 @param nodecert_file: optional override of the node cert file path
127 @type rapicert_file: string
128 @param rapicert_file: optional override of the rapi cert file path
129 @type spicecert_file: string
130 @param spicecert_file: optional override of the spice cert file path
131 @type spicecacert_file: string
132 @param spicecacert_file: optional override of the spice CA cert file path
133 @type hmackey_file: string
134 @param hmackey_file: optional override of the hmac key file path
137 # noded SSL certificate
138 cluster_cert_exists = os.path.exists(nodecert_file)
139 if new_cluster_cert or not cluster_cert_exists:
140 if cluster_cert_exists:
141 utils.CreateBackup(nodecert_file)
143 logging.debug("Generating new cluster certificate at %s", nodecert_file)
144 utils.GenerateSelfSignedSslCert(nodecert_file)
147 if new_confd_hmac_key or not os.path.exists(hmackey_file):
148 logging.debug("Writing new confd HMAC key to %s", hmackey_file)
149 GenerateHmacKey(hmackey_file)
152 rapi_cert_exists = os.path.exists(rapicert_file)
155 # Assume rapi_pem contains a valid PEM-formatted certificate and key
156 logging.debug("Writing RAPI certificate at %s", rapicert_file)
157 utils.WriteFile(rapicert_file, data=rapi_cert_pem, backup=True)
159 elif new_rapi_cert or not rapi_cert_exists:
161 utils.CreateBackup(rapicert_file)
163 logging.debug("Generating new RAPI certificate at %s", rapicert_file)
164 utils.GenerateSelfSignedSslCert(rapicert_file)
167 spice_cert_exists = os.path.exists(spicecert_file)
168 spice_cacert_exists = os.path.exists(spicecacert_file)
170 # spice_cert_pem implies also spice_cacert_pem
171 logging.debug("Writing SPICE certificate at %s", spicecert_file)
172 utils.WriteFile(spicecert_file, data=spice_cert_pem, backup=True)
173 logging.debug("Writing SPICE CA certificate at %s", spicecacert_file)
174 utils.WriteFile(spicecacert_file, data=spice_cacert_pem, backup=True)
175 elif new_spice_cert or not spice_cert_exists:
176 if spice_cert_exists:
177 utils.CreateBackup(spicecert_file)
178 if spice_cacert_exists:
179 utils.CreateBackup(spicecacert_file)
181 logging.debug("Generating new self-signed SPICE certificate at %s",
183 (_, cert_pem) = utils.GenerateSelfSignedSslCert(spicecert_file)
185 # Self-signed certificate -> the public certificate is also the CA public
187 logging.debug("Writing the public certificate to %s",
189 utils.io.WriteFile(spicecacert_file, mode=0400, data=cert_pem)
191 # Cluster domain secret
193 logging.debug("Writing cluster domain secret to %s", cds_file)
194 utils.WriteFile(cds_file, data=cds, backup=True)
196 elif new_cds or not os.path.exists(cds_file):
197 logging.debug("Generating new cluster domain secret at %s", cds_file)
198 GenerateHmacKey(cds_file)
201 def _InitGanetiServerSetup(master_name):
202 """Setup the necessary configuration for the initial node daemon.
204 This creates the nodepass file containing the shared password for
205 the cluster, generates the SSL certificate and starts the node daemon.
207 @type master_name: str
208 @param master_name: Name of the master node
211 # Generate cluster secrets
212 GenerateClusterCrypto(True, False, False, False, False)
214 result = utils.RunCmd([pathutils.DAEMON_UTIL, "start", constants.NODED])
216 raise errors.OpExecError("Could not start the node daemon, command %s"
217 " had exitcode %s and error %s" %
218 (result.cmd, result.exit_code, result.output))
220 _WaitForNodeDaemon(master_name)
223 def _WaitForNodeDaemon(node_name):
224 """Wait for node daemon to become responsive.
227 def _CheckNodeDaemon():
228 # Pylint bug <http://www.logilab.org/ticket/35642>
229 # pylint: disable=E1101
230 result = rpc.BootstrapRunner().call_version([node_name])[node_name]
232 raise utils.RetryAgain()
235 utils.Retry(_CheckNodeDaemon, 1.0, _DAEMON_READY_TIMEOUT)
236 except utils.RetryTimeout:
237 raise errors.OpExecError("Node daemon on %s didn't answer queries within"
238 " %s seconds" % (node_name, _DAEMON_READY_TIMEOUT))
241 def _WaitForMasterDaemon():
242 """Wait for master daemon to become responsive.
245 def _CheckMasterDaemon():
248 (cluster_name, ) = cl.QueryConfigValues(["cluster_name"])
250 raise utils.RetryAgain()
252 logging.debug("Received cluster name %s from master", cluster_name)
255 utils.Retry(_CheckMasterDaemon, 1.0, _DAEMON_READY_TIMEOUT)
256 except utils.RetryTimeout:
257 raise errors.OpExecError("Master daemon didn't answer queries within"
258 " %s seconds" % _DAEMON_READY_TIMEOUT)
261 def _WaitForSshDaemon(hostname, port, family):
262 """Wait for SSH daemon to become responsive.
265 hostip = netutils.GetHostname(name=hostname, family=family).ip
267 def _CheckSshDaemon():
268 if netutils.TcpPing(hostip, port, timeout=1.0, live_port_needed=True):
269 logging.debug("SSH daemon on %s:%s (IP address %s) has become"
270 " responsive", hostname, port, hostip)
272 raise utils.RetryAgain()
275 utils.Retry(_CheckSshDaemon, 1.0, _DAEMON_READY_TIMEOUT)
276 except utils.RetryTimeout:
277 raise errors.OpExecError("SSH daemon on %s:%s (IP address %s) didn't"
278 " become responsive within %s seconds" %
279 (hostname, port, hostip, _DAEMON_READY_TIMEOUT))
282 def RunNodeSetupCmd(cluster_name, node, basecmd, debug, verbose,
283 use_cluster_key, ask_key, strict_host_check, data):
284 """Runs a command to configure something on a remote machine.
286 @type cluster_name: string
287 @param cluster_name: Cluster name
289 @param node: Node name
290 @type basecmd: string
291 @param basecmd: Base command (path on the remote machine)
293 @param debug: Enable debug output
295 @param verbose: Enable verbose output
296 @type use_cluster_key: bool
297 @param use_cluster_key: See L{ssh.SshRunner.BuildCmd}
299 @param ask_key: See L{ssh.SshRunner.BuildCmd}
300 @type strict_host_check: bool
301 @param strict_host_check: See L{ssh.SshRunner.BuildCmd}
302 @param data: JSON-serializable input data for script (passed to stdin)
307 # Pass --debug/--verbose to the external script if set on our invocation
309 cmd.append("--debug")
312 cmd.append("--verbose")
314 logging.debug("Node setup command: %s", cmd)
316 version = constants.DIR_VERSION
317 all_cmds = [["test", "-d", os.path.join(pathutils.PKGLIBDIR, version)]]
318 if constants.HAS_GNU_LN:
319 all_cmds.extend([["ln", "-s", "-f", "-T",
320 os.path.join(pathutils.PKGLIBDIR, version),
321 os.path.join(pathutils.SYSCONFDIR, "ganeti/lib")],
322 ["ln", "-s", "-f", "-T",
323 os.path.join(pathutils.SHAREDIR, version),
324 os.path.join(pathutils.SYSCONFDIR, "ganeti/share")]])
326 all_cmds.extend([["rm", "-f",
327 os.path.join(pathutils.SYSCONFDIR, "ganeti/lib")],
329 os.path.join(pathutils.PKGLIBDIR, version),
330 os.path.join(pathutils.SYSCONFDIR, "ganeti/lib")],
332 os.path.join(pathutils.SYSCONFDIR, "ganeti/share")],
334 os.path.join(pathutils.SHAREDIR, version),
335 os.path.join(pathutils.SYSCONFDIR, "ganeti/share")]])
338 family = ssconf.SimpleStore().GetPrimaryIPFamily()
339 srun = ssh.SshRunner(cluster_name,
340 ipv6=(family == netutils.IP6Address.family))
341 scmd = srun.BuildCmd(node, constants.SSH_LOGIN_USER,
342 utils.ShellQuoteArgs(
343 utils.ShellCombineCommands(all_cmds)),
344 batch=False, ask_key=ask_key, quiet=False,
345 strict_host_check=strict_host_check,
346 use_cluster_key=use_cluster_key)
348 tempfh = tempfile.TemporaryFile()
350 tempfh.write(serializer.DumpJson(data))
353 result = utils.RunCmd(scmd, interactive=True, input_fd=tempfh)
358 raise errors.OpExecError("Command '%s' failed: %s" %
359 (result.cmd, result.fail_reason))
361 _WaitForSshDaemon(node, netutils.GetDaemonPort(constants.SSH), family)
364 def _InitFileStorageDir(file_storage_dir):
365 """Initialize if needed the file storage.
367 @param file_storage_dir: the user-supplied value
368 @return: either empty string (if file storage was disabled at build
369 time) or the normalized path to the storage directory
372 file_storage_dir = os.path.normpath(file_storage_dir)
374 if not os.path.isabs(file_storage_dir):
375 raise errors.OpPrereqError("File storage directory '%s' is not an absolute"
376 " path" % file_storage_dir, errors.ECODE_INVAL)
378 if not os.path.exists(file_storage_dir):
380 os.makedirs(file_storage_dir, 0750)
382 raise errors.OpPrereqError("Cannot create file storage directory"
383 " '%s': %s" % (file_storage_dir, err),
384 errors.ECODE_ENVIRON)
386 if not os.path.isdir(file_storage_dir):
387 raise errors.OpPrereqError("The file storage directory '%s' is not"
388 " a directory." % file_storage_dir,
389 errors.ECODE_ENVIRON)
391 return file_storage_dir
394 def _PrepareFileBasedStorage(
395 enabled_disk_templates, file_storage_dir,
396 default_dir, file_disk_template,
397 init_fn=_InitFileStorageDir, acceptance_fn=None):
398 """Checks if a file-base storage type is enabled and inits the dir.
400 @type enabled_disk_templates: list of string
401 @param enabled_disk_templates: list of enabled disk templates
402 @type file_storage_dir: string
403 @param file_storage_dir: the file storage directory
404 @type default_dir: string
405 @param default_dir: default file storage directory when C{file_storage_dir}
407 @type file_disk_template: string
408 @param file_disk_template: a disk template whose storage type is 'ST_FILE'
410 @returns: the name of the actual file storage directory
413 assert (file_disk_template in
414 utils.storage.GetDiskTemplatesOfStorageType(constants.ST_FILE))
415 if file_storage_dir is None:
416 file_storage_dir = default_dir
417 if not acceptance_fn:
419 lambda path: filestorage.CheckFileStoragePathAcceptance(
420 path, exact_match_ok=True)
422 cluster.CheckFileStoragePathVsEnabledDiskTemplates(
423 logging.warning, file_storage_dir, enabled_disk_templates)
425 file_storage_enabled = file_disk_template in enabled_disk_templates
426 if file_storage_enabled:
428 acceptance_fn(file_storage_dir)
429 except errors.FileStoragePathError as e:
430 raise errors.OpPrereqError(str(e))
431 result_file_storage_dir = init_fn(file_storage_dir)
433 result_file_storage_dir = file_storage_dir
434 return result_file_storage_dir
437 def _PrepareFileStorage(
438 enabled_disk_templates, file_storage_dir, init_fn=_InitFileStorageDir,
440 """Checks if file storage is enabled and inits the dir.
442 @see: C{_PrepareFileBasedStorage}
445 return _PrepareFileBasedStorage(
446 enabled_disk_templates, file_storage_dir,
447 pathutils.DEFAULT_FILE_STORAGE_DIR, constants.DT_FILE,
448 init_fn=init_fn, acceptance_fn=acceptance_fn)
451 def _PrepareSharedFileStorage(
452 enabled_disk_templates, file_storage_dir, init_fn=_InitFileStorageDir,
454 """Checks if shared file storage is enabled and inits the dir.
456 @see: C{_PrepareFileBasedStorage}
459 return _PrepareFileBasedStorage(
460 enabled_disk_templates, file_storage_dir,
461 pathutils.DEFAULT_SHARED_FILE_STORAGE_DIR, constants.DT_SHARED_FILE,
462 init_fn=init_fn, acceptance_fn=acceptance_fn)
465 def _InitCheckEnabledDiskTemplates(enabled_disk_templates):
466 """Checks the sanity of the enabled disk templates.
469 if not enabled_disk_templates:
470 raise errors.OpPrereqError("Enabled disk templates list must contain at"
471 " least one member", errors.ECODE_INVAL)
472 invalid_disk_templates = \
473 set(enabled_disk_templates) - constants.DISK_TEMPLATES
474 if invalid_disk_templates:
475 raise errors.OpPrereqError("Enabled disk templates list contains invalid"
476 " entries: %s" % invalid_disk_templates,
480 def _RestrictIpolicyToEnabledDiskTemplates(ipolicy, enabled_disk_templates):
481 """Restricts the ipolicy's disk templates to the enabled ones.
483 This function clears the ipolicy's list of allowed disk templates from the
484 ones that are not enabled by the cluster.
487 @param ipolicy: the instance policy
488 @type enabled_disk_templates: list of string
489 @param enabled_disk_templates: the list of cluster-wide enabled disk
493 assert constants.IPOLICY_DTS in ipolicy
494 allowed_disk_templates = ipolicy[constants.IPOLICY_DTS]
495 restricted_disk_templates = list(set(allowed_disk_templates)
496 .intersection(set(enabled_disk_templates)))
497 ipolicy[constants.IPOLICY_DTS] = restricted_disk_templates
500 def _InitCheckDrbdHelper(drbd_helper, drbd_enabled):
501 """Checks the DRBD usermode helper.
503 @type drbd_helper: string
504 @param drbd_helper: name of the DRBD usermode helper that the system should
511 if drbd_helper is not None:
513 curr_helper = drbd.DRBD8.GetUsermodeHelper()
514 except errors.BlockDeviceError, err:
515 raise errors.OpPrereqError("Error while checking drbd helper"
516 " (disable drbd with --enabled-disk-templates"
517 " if you are not using drbd): %s" % str(err),
518 errors.ECODE_ENVIRON)
519 if drbd_helper != curr_helper:
520 raise errors.OpPrereqError("Error: requiring %s as drbd helper but %s"
521 " is the current helper" % (drbd_helper,
526 def InitCluster(cluster_name, mac_prefix, # pylint: disable=R0913, R0914
527 master_netmask, master_netdev, file_storage_dir,
528 shared_file_storage_dir, candidate_pool_size, secondary_ip=None,
529 vg_name=None, beparams=None, nicparams=None, ndparams=None,
530 hvparams=None, diskparams=None, enabled_hypervisors=None,
531 modify_etc_hosts=True, modify_ssh_setup=True,
532 maintain_node_health=False, drbd_helper=None, uid_pool=None,
533 default_iallocator=None, primary_ip_version=None, ipolicy=None,
534 prealloc_wipe_disks=False, use_external_mip_script=False,
535 hv_state=None, disk_state=None, enabled_disk_templates=None):
536 """Initialise the cluster.
538 @type candidate_pool_size: int
539 @param candidate_pool_size: master candidate pool size
540 @type enabled_disk_templates: list of string
541 @param enabled_disk_templates: list of disk_templates to be used in this
545 # TODO: complete the docstring
546 if config.ConfigWriter.IsCluster():
547 raise errors.OpPrereqError("Cluster is already initialised",
550 if not enabled_hypervisors:
551 raise errors.OpPrereqError("Enabled hypervisors list must contain at"
552 " least one member", errors.ECODE_INVAL)
553 invalid_hvs = set(enabled_hypervisors) - constants.HYPER_TYPES
555 raise errors.OpPrereqError("Enabled hypervisors contains invalid"
556 " entries: %s" % invalid_hvs,
559 _InitCheckEnabledDiskTemplates(enabled_disk_templates)
562 ipcls = netutils.IPAddress.GetClassFromIpVersion(primary_ip_version)
563 except errors.ProgrammerError:
564 raise errors.OpPrereqError("Invalid primary ip version: %d." %
565 primary_ip_version, errors.ECODE_INVAL)
567 hostname = netutils.GetHostname(family=ipcls.family)
568 if not ipcls.IsValid(hostname.ip):
569 raise errors.OpPrereqError("This host's IP (%s) is not a valid IPv%d"
570 " address." % (hostname.ip, primary_ip_version),
573 if ipcls.IsLoopback(hostname.ip):
574 raise errors.OpPrereqError("This host's IP (%s) resolves to a loopback"
575 " address. Please fix DNS or %s." %
576 (hostname.ip, pathutils.ETC_HOSTS),
577 errors.ECODE_ENVIRON)
579 if not ipcls.Own(hostname.ip):
580 raise errors.OpPrereqError("Inconsistency: this host's name resolves"
581 " to %s,\nbut this ip address does not"
582 " belong to this host" %
583 hostname.ip, errors.ECODE_ENVIRON)
585 clustername = netutils.GetHostname(name=cluster_name, family=ipcls.family)
587 if netutils.TcpPing(clustername.ip, constants.DEFAULT_NODED_PORT, timeout=5):
588 raise errors.OpPrereqError("Cluster IP already active",
589 errors.ECODE_NOTUNIQUE)
592 if primary_ip_version == constants.IP6_VERSION:
593 raise errors.OpPrereqError("When using a IPv6 primary address, a valid"
594 " IPv4 address must be given as secondary",
596 secondary_ip = hostname.ip
598 if not netutils.IP4Address.IsValid(secondary_ip):
599 raise errors.OpPrereqError("Secondary IP address (%s) has to be a valid"
600 " IPv4 address." % secondary_ip,
603 if not netutils.IP4Address.Own(secondary_ip):
604 raise errors.OpPrereqError("You gave %s as secondary IP,"
605 " but it does not belong to this host." %
606 secondary_ip, errors.ECODE_ENVIRON)
608 if master_netmask is not None:
609 if not ipcls.ValidateNetmask(master_netmask):
610 raise errors.OpPrereqError("CIDR netmask (%s) not valid for IPv%s " %
611 (master_netmask, primary_ip_version),
614 master_netmask = ipcls.iplen
617 # Check if volume group is valid
618 vgstatus = utils.CheckVolumeGroupSize(utils.ListVolumeGroups(), vg_name,
619 constants.MIN_VG_SIZE)
621 raise errors.OpPrereqError("Error: %s" % vgstatus, errors.ECODE_INVAL)
623 drbd_enabled = constants.DT_DRBD8 in enabled_disk_templates
624 _InitCheckDrbdHelper(drbd_helper, drbd_enabled)
626 logging.debug("Stopping daemons (if any are running)")
627 result = utils.RunCmd([pathutils.DAEMON_UTIL, "stop-all"])
629 raise errors.OpExecError("Could not stop daemons, command %s"
630 " had exitcode %s and error '%s'" %
631 (result.cmd, result.exit_code, result.output))
633 file_storage_dir = _PrepareFileStorage(enabled_disk_templates,
635 shared_file_storage_dir = _PrepareSharedFileStorage(enabled_disk_templates,
636 shared_file_storage_dir)
638 if not re.match("^[0-9a-z]{2}:[0-9a-z]{2}:[0-9a-z]{2}$", mac_prefix):
639 raise errors.OpPrereqError("Invalid mac prefix given '%s'" % mac_prefix,
642 if not nicparams.get('mode', None) == constants.NIC_MODE_OVS:
643 # Do not do this check if mode=openvswitch, since the openvswitch is not
645 result = utils.RunCmd(["ip", "link", "show", "dev", master_netdev])
647 raise errors.OpPrereqError("Invalid master netdev given (%s): '%s'" %
649 result.output.strip()), errors.ECODE_INVAL)
651 dirs = [(pathutils.RUN_DIR, constants.RUN_DIRS_MODE)]
652 utils.EnsureDirs(dirs)
654 objects.UpgradeBeParams(beparams)
655 utils.ForceDictType(beparams, constants.BES_PARAMETER_TYPES)
656 utils.ForceDictType(nicparams, constants.NICS_PARAMETER_TYPES)
658 objects.NIC.CheckParameterSyntax(nicparams)
660 full_ipolicy = objects.FillIPolicy(constants.IPOLICY_DEFAULTS, ipolicy)
661 _RestrictIpolicyToEnabledDiskTemplates(full_ipolicy, enabled_disk_templates)
663 if ndparams is not None:
664 utils.ForceDictType(ndparams, constants.NDS_PARAMETER_TYPES)
666 ndparams = dict(constants.NDC_DEFAULTS)
668 # This is ugly, as we modify the dict itself
669 # FIXME: Make utils.ForceDictType pure functional or write a wrapper
672 for hvname, hvs_data in hv_state.items():
673 utils.ForceDictType(hvs_data, constants.HVSTS_PARAMETER_TYPES)
674 hv_state[hvname] = objects.Cluster.SimpleFillHvState(hvs_data)
676 hv_state = dict((hvname, constants.HVST_DEFAULTS)
677 for hvname in enabled_hypervisors)
679 # FIXME: disk_state has no default values yet
681 for storage, ds_data in disk_state.items():
682 if storage not in constants.DS_VALID_TYPES:
683 raise errors.OpPrereqError("Invalid storage type in disk state: %s" %
684 storage, errors.ECODE_INVAL)
685 for ds_name, state in ds_data.items():
686 utils.ForceDictType(state, constants.DSS_PARAMETER_TYPES)
687 ds_data[ds_name] = objects.Cluster.SimpleFillDiskState(state)
689 # hvparams is a mapping of hypervisor->hvparams dict
690 for hv_name, hv_params in hvparams.iteritems():
691 utils.ForceDictType(hv_params, constants.HVS_PARAMETER_TYPES)
692 hv_class = hypervisor.GetHypervisor(hv_name)
693 hv_class.CheckParameterSyntax(hv_params)
695 # diskparams is a mapping of disk-template->diskparams dict
696 for template, dt_params in diskparams.items():
697 param_keys = set(dt_params.keys())
698 default_param_keys = set(constants.DISK_DT_DEFAULTS[template].keys())
699 if not (param_keys <= default_param_keys):
700 unknown_params = param_keys - default_param_keys
701 raise errors.OpPrereqError("Invalid parameters for disk template %s:"
703 utils.CommaJoin(unknown_params)),
705 utils.ForceDictType(dt_params, constants.DISK_DT_TYPES)
706 if template == constants.DT_DRBD8 and vg_name is not None:
707 # The default METAVG value is equal to the VG name set at init time,
709 dt_params[constants.DRBD_DEFAULT_METAVG] = vg_name
712 utils.VerifyDictOptions(diskparams, constants.DISK_DT_DEFAULTS)
713 except errors.OpPrereqError, err:
714 raise errors.OpPrereqError("While verify diskparam options: %s" % err,
717 # set up ssh config and /etc/hosts
720 if os.path.isfile(pathutils.SSH_HOST_RSA_PUB):
721 sshline = utils.ReadFile(pathutils.SSH_HOST_RSA_PUB)
722 rsa_sshkey = sshline.split(" ")[1]
723 if os.path.isfile(pathutils.SSH_HOST_DSA_PUB):
724 sshline = utils.ReadFile(pathutils.SSH_HOST_DSA_PUB)
725 dsa_sshkey = sshline.split(" ")[1]
726 if not rsa_sshkey and not dsa_sshkey:
727 raise errors.OpPrereqError("Failed to find SSH public keys",
728 errors.ECODE_ENVIRON)
731 utils.AddHostToEtcHosts(hostname.name, hostname.ip)
736 if default_iallocator is not None:
737 alloc_script = utils.FindFile(default_iallocator,
738 constants.IALLOCATOR_SEARCH_PATH,
740 if alloc_script is None:
741 raise errors.OpPrereqError("Invalid default iallocator script '%s'"
742 " specified" % default_iallocator,
744 elif constants.HTOOLS:
745 # htools was enabled at build-time, we default to it
746 if utils.FindFile(constants.IALLOC_HAIL,
747 constants.IALLOCATOR_SEARCH_PATH,
749 default_iallocator = constants.IALLOC_HAIL
753 # init of cluster config file
754 cluster_config = objects.Cluster(
756 rsahostkeypub=rsa_sshkey,
757 dsahostkeypub=dsa_sshkey,
758 highest_used_port=(constants.FIRST_DRBD_PORT - 1),
759 mac_prefix=mac_prefix,
760 volume_group_name=vg_name,
761 tcpudp_port_pool=set(),
762 master_ip=clustername.ip,
763 master_netmask=master_netmask,
764 master_netdev=master_netdev,
765 cluster_name=clustername.name,
766 file_storage_dir=file_storage_dir,
767 shared_file_storage_dir=shared_file_storage_dir,
768 enabled_hypervisors=enabled_hypervisors,
769 beparams={constants.PP_DEFAULT: beparams},
770 nicparams={constants.PP_DEFAULT: nicparams},
773 diskparams=diskparams,
774 candidate_pool_size=candidate_pool_size,
775 modify_etc_hosts=modify_etc_hosts,
776 modify_ssh_setup=modify_ssh_setup,
780 maintain_node_health=maintain_node_health,
781 drbd_usermode_helper=drbd_helper,
782 default_iallocator=default_iallocator,
783 primary_ip_family=ipcls.family,
784 prealloc_wipe_disks=prealloc_wipe_disks,
785 use_external_mip_script=use_external_mip_script,
786 ipolicy=full_ipolicy,
787 hv_state_static=hv_state,
788 disk_state_static=disk_state,
789 enabled_disk_templates=enabled_disk_templates,
791 master_node_config = objects.Node(name=hostname.name,
792 primary_ip=hostname.ip,
793 secondary_ip=secondary_ip,
795 master_candidate=True,
796 offline=False, drained=False,
797 ctime=now, mtime=now,
799 InitConfig(constants.CONFIG_VERSION, cluster_config, master_node_config)
800 cfg = config.ConfigWriter(offline=True)
801 ssh.WriteKnownHostsFile(cfg, pathutils.SSH_KNOWN_HOSTS_FILE)
802 cfg.Update(cfg.GetClusterInfo(), logging.error)
803 ssconf.WriteSsconfFiles(cfg.GetSsconfValues())
805 # set up the inter-node password and certificate
806 _InitGanetiServerSetup(hostname.name)
808 logging.debug("Starting daemons")
809 result = utils.RunCmd([pathutils.DAEMON_UTIL, "start-all"])
811 raise errors.OpExecError("Could not start daemons, command %s"
812 " had exitcode %s and error %s" %
813 (result.cmd, result.exit_code, result.output))
815 _WaitForMasterDaemon()
818 def InitConfig(version, cluster_config, master_node_config,
819 cfg_file=pathutils.CLUSTER_CONF_FILE):
820 """Create the initial cluster configuration.
822 It will contain the current node, which will also be the master
823 node, and no instances.
826 @param version: configuration version
827 @type cluster_config: L{objects.Cluster}
828 @param cluster_config: cluster configuration
829 @type master_node_config: L{objects.Node}
830 @param master_node_config: master node configuration
831 @type cfg_file: string
832 @param cfg_file: configuration file path
835 uuid_generator = config.TemporaryReservationManager()
836 cluster_config.uuid = uuid_generator.Generate([], utils.NewUUID,
838 master_node_config.uuid = uuid_generator.Generate([], utils.NewUUID,
840 cluster_config.master_node = master_node_config.uuid
842 master_node_config.uuid: master_node_config,
844 default_nodegroup = objects.NodeGroup(
845 uuid=uuid_generator.Generate([], utils.NewUUID, _INITCONF_ECID),
846 name=constants.INITIAL_NODE_GROUP_NAME,
847 members=[master_node_config.uuid],
851 default_nodegroup.uuid: default_nodegroup,
854 config_data = objects.ConfigData(version=version,
855 cluster=cluster_config,
856 nodegroups=nodegroups,
861 ctime=now, mtime=now)
862 utils.WriteFile(cfg_file,
863 data=serializer.Dump(config_data.ToDict()),
867 def FinalizeClusterDestroy(master_uuid):
868 """Execute the last steps of cluster destroy
870 This function shuts down all the daemons, completing the destroy
871 begun in cmdlib.LUDestroyOpcode.
874 cfg = config.ConfigWriter()
875 modify_ssh_setup = cfg.GetClusterInfo().modify_ssh_setup
876 runner = rpc.BootstrapRunner()
878 master_name = cfg.GetNodeName(master_uuid)
880 master_params = cfg.GetMasterNetworkParameters()
881 master_params.uuid = master_uuid
882 ems = cfg.GetUseExternalMipScript()
883 result = runner.call_node_deactivate_master_ip(master_name, master_params,
886 msg = result.fail_msg
888 logging.warning("Could not disable the master IP: %s", msg)
890 result = runner.call_node_stop_master(master_name)
891 msg = result.fail_msg
893 logging.warning("Could not disable the master role: %s", msg)
895 result = runner.call_node_leave_cluster(master_name, modify_ssh_setup)
896 msg = result.fail_msg
898 logging.warning("Could not shutdown the node daemon and cleanup"
899 " the node: %s", msg)
902 def SetupNodeDaemon(opts, cluster_name, node):
903 """Add a node to the cluster.
905 This function must be called before the actual opcode, and will ssh
906 to the remote node, copy the needed files, and start ganeti-noded,
907 allowing the master to do the rest via normal rpc calls.
909 @param cluster_name: the cluster name
910 @param node: the name of the new node
914 constants.NDS_CLUSTER_NAME: cluster_name,
915 constants.NDS_NODE_DAEMON_CERTIFICATE:
916 utils.ReadFile(pathutils.NODED_CERT_FILE),
917 constants.NDS_SSCONF: ssconf.SimpleStore().ReadAll(),
918 constants.NDS_START_NODE_DAEMON: True,
921 RunNodeSetupCmd(cluster_name, node, pathutils.NODE_DAEMON_SETUP,
922 opts.debug, opts.verbose,
923 True, opts.ssh_key_check, opts.ssh_key_check, data)
925 _WaitForNodeDaemon(node)
928 def MasterFailover(no_voting=False):
929 """Failover the master node.
931 This checks that we are not already the master, and will cause the
932 current master to cease being master, and the non-master to become
935 @type no_voting: boolean
936 @param no_voting: force the operation without remote nodes agreement
940 sstore = ssconf.SimpleStore()
942 old_master, new_master = ssconf.GetMasterAndMyself(sstore)
943 node_names = sstore.GetNodeList()
944 mc_list = sstore.GetMasterCandidates()
946 if old_master == new_master:
947 raise errors.OpPrereqError("This commands must be run on the node"
948 " where you want the new master to be."
949 " %s is already the master" %
950 old_master, errors.ECODE_INVAL)
952 if new_master not in mc_list:
953 mc_no_master = [name for name in mc_list if name != old_master]
954 raise errors.OpPrereqError("This node is not among the nodes marked"
955 " as master candidates. Only these nodes"
956 " can become masters. Current list of"
957 " master candidates is:\n"
958 "%s" % ("\n".join(mc_no_master)),
962 vote_list = GatherMasterVotes(node_names)
965 voted_master = vote_list[0][0]
966 if voted_master is None:
967 raise errors.OpPrereqError("Cluster is inconsistent, most nodes did"
968 " not respond.", errors.ECODE_ENVIRON)
969 elif voted_master != old_master:
970 raise errors.OpPrereqError("I have a wrong configuration, I believe"
971 " the master is %s but the other nodes"
972 " voted %s. Please resync the configuration"
974 (old_master, voted_master),
980 logging.info("Setting master to %s, old master: %s", new_master, old_master)
983 # instantiate a real config writer, as we now know we have the
985 cfg = config.ConfigWriter(accept_foreign=True)
987 old_master_node = cfg.GetNodeInfoByName(old_master)
988 if old_master_node is None:
989 raise errors.OpPrereqError("Could not find old master node '%s' in"
990 " cluster configuration." % old_master,
993 cluster_info = cfg.GetClusterInfo()
994 new_master_node = cfg.GetNodeInfoByName(new_master)
995 if new_master_node is None:
996 raise errors.OpPrereqError("Could not find new master node '%s' in"
997 " cluster configuration." % new_master,
1000 cluster_info.master_node = new_master_node.uuid
1001 # this will also regenerate the ssconf files, since we updated the
1003 cfg.Update(cluster_info, logging.error)
1004 except errors.ConfigurationError, err:
1005 logging.error("Error while trying to set the new master: %s",
1009 # if cfg.Update worked, then it means the old master daemon won't be
1010 # able now to write its own config file (we rely on locking in both
1011 # backend.UploadFile() and ConfigWriter._Write(); hence the next
1012 # step is to kill the old master
1014 logging.info("Stopping the master daemon on node %s", old_master)
1016 runner = rpc.BootstrapRunner()
1017 master_params = cfg.GetMasterNetworkParameters()
1018 master_params.uuid = old_master_node.uuid
1019 ems = cfg.GetUseExternalMipScript()
1020 result = runner.call_node_deactivate_master_ip(old_master,
1023 msg = result.fail_msg
1025 logging.warning("Could not disable the master IP: %s", msg)
1027 result = runner.call_node_stop_master(old_master)
1028 msg = result.fail_msg
1030 logging.error("Could not disable the master role on the old master"
1031 " %s, please disable manually: %s", old_master, msg)
1033 logging.info("Checking master IP non-reachability...")
1035 master_ip = sstore.GetMasterIP()
1038 # Here we have a phase where no master should be running
1040 if netutils.TcpPing(master_ip, constants.DEFAULT_NODED_PORT):
1041 raise utils.RetryAgain()
1044 utils.Retry(_check_ip, (1, 1.5, 5), total_timeout)
1045 except utils.RetryTimeout:
1046 logging.warning("The master IP is still reachable after %s seconds,"
1047 " continuing but activating the master on the current"
1048 " node will probably fail", total_timeout)
1050 if jstore.CheckDrainFlag():
1051 logging.info("Undraining job queue")
1052 jstore.SetDrainFlag(False)
1054 logging.info("Starting the master daemons on the new master")
1056 result = rpc.BootstrapRunner().call_node_start_master_daemons(new_master,
1058 msg = result.fail_msg
1060 logging.error("Could not start the master role on the new master"
1061 " %s, please check: %s", new_master, msg)
1064 logging.info("Master failed over from %s to %s", old_master, new_master)
1069 """Returns the current master node.
1071 This is a separate function in bootstrap since it's needed by
1072 gnt-cluster, and instead of importing directly ssconf, it's better
1073 to abstract it in bootstrap, where we do use ssconf in other
1077 sstore = ssconf.SimpleStore()
1079 old_master, _ = ssconf.GetMasterAndMyself(sstore)
1084 def GatherMasterVotes(node_names):
1085 """Check the agreement on who is the master.
1087 This function will return a list of (node, number of votes), ordered
1088 by the number of votes. Errors will be denoted by the key 'None'.
1090 Note that the sum of votes is the number of nodes this machine
1091 knows, whereas the number of entries in the list could be different
1092 (if some nodes vote for another master).
1094 We remove ourselves from the list since we know that (bugs aside)
1095 since we use the same source for configuration information for both
1096 backend and boostrap, we'll always vote for ourselves.
1098 @type node_names: list
1099 @param node_names: the list of nodes to query for master info; the current
1100 node will be removed if it is in the list
1102 @return: list of (node, votes)
1105 myself = netutils.Hostname.GetSysName()
1107 node_names.remove(myself)
1111 # no nodes left (eventually after removing myself)
1113 results = rpc.BootstrapRunner().call_master_info(node_names)
1114 if not isinstance(results, dict):
1115 # this should not happen (unless internal error in rpc)
1116 logging.critical("Can't complete rpc call, aborting master startup")
1117 return [(None, len(node_names))]
1119 for node_name in results:
1120 nres = results[node_name]
1125 logging.warning("Error contacting node %s: %s", node_name, msg)
1127 # for now we accept both length 3, 4 and 5 (data[3] is primary ip version
1128 # and data[4] is the master netmask)
1129 elif not isinstance(data, (tuple, list)) or len(data) < 3:
1130 logging.warning("Invalid data received from node %s: %s",
1134 if None not in votes:
1138 master_node = data[2]
1139 if master_node not in votes:
1140 votes[master_node] = 0
1141 votes[master_node] += 1
1143 vote_list = [v for v in votes.items()]
1144 # sort first on number of votes then on name, since we want None
1145 # sorted later if we have the half of the nodes not responding, and
1146 # half voting all for the same master
1147 vote_list.sort(key=lambda x: (x[1], x[0]), reverse=True)