Statistics
| Branch: | Tag: | Revision:

root / lib / bootstrap.py @ 5030cff3

History | View | Annotate | Download (37.4 kB)

1
#
2
#
3

    
4
# Copyright (C) 2006, 2007, 2008, 2010, 2011, 2012 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
"""Functions to bootstrap a new cluster.
23

24
"""
25

    
26
import os
27
import os.path
28
import re
29
import logging
30
import time
31
import tempfile
32

    
33
from ganeti import rpc
34
from ganeti import ssh
35
from ganeti import utils
36
from ganeti import errors
37
from ganeti import config
38
from ganeti import constants
39
from ganeti import objects
40
from ganeti import ssconf
41
from ganeti import serializer
42
from ganeti import hypervisor
43
from ganeti.storage import drbd
44
from ganeti import netutils
45
from ganeti import luxi
46
from ganeti import jstore
47
from ganeti import pathutils
48

    
49

    
50
# ec_id for InitConfig's temporary reservation manager
51
_INITCONF_ECID = "initconfig-ecid"
52

    
53
#: After how many seconds daemon must be responsive
54
_DAEMON_READY_TIMEOUT = 10.0
55

    
56

    
57
def _InitSSHSetup():
58
  """Setup the SSH configuration for the cluster.
59

60
  This generates a dsa keypair for root, adds the pub key to the
61
  permitted hosts and adds the hostkey to its own known hosts.
62

63
  """
64
  priv_key, pub_key, auth_keys = ssh.GetUserFiles(constants.SSH_LOGIN_USER)
65

    
66
  for name in priv_key, pub_key:
67
    if os.path.exists(name):
68
      utils.CreateBackup(name)
69
    utils.RemoveFile(name)
70

    
71
  result = utils.RunCmd(["ssh-keygen", "-t", "dsa",
72
                         "-f", priv_key,
73
                         "-q", "-N", ""])
74
  if result.failed:
75
    raise errors.OpExecError("Could not generate ssh keypair, error %s" %
76
                             result.output)
77

    
78
  utils.AddAuthorizedKey(auth_keys, utils.ReadFile(pub_key))
79

    
80

    
81
def GenerateHmacKey(file_name):
82
  """Writes a new HMAC key.
83

84
  @type file_name: str
85
  @param file_name: Path to output file
86

87
  """
88
  utils.WriteFile(file_name, data="%s\n" % utils.GenerateSecret(), mode=0400,
89
                  backup=True)
90

    
91

    
92
def GenerateClusterCrypto(new_cluster_cert, new_rapi_cert, new_spice_cert,
93
                          new_confd_hmac_key, new_cds,
94
                          rapi_cert_pem=None, spice_cert_pem=None,
95
                          spice_cacert_pem=None, cds=None,
96
                          nodecert_file=pathutils.NODED_CERT_FILE,
97
                          rapicert_file=pathutils.RAPI_CERT_FILE,
98
                          spicecert_file=pathutils.SPICE_CERT_FILE,
99
                          spicecacert_file=pathutils.SPICE_CACERT_FILE,
100
                          hmackey_file=pathutils.CONFD_HMAC_KEY,
101
                          cds_file=pathutils.CLUSTER_DOMAIN_SECRET_FILE):
102
  """Updates the cluster certificates, keys and secrets.
103

104
  @type new_cluster_cert: bool
105
  @param new_cluster_cert: Whether to generate a new cluster certificate
106
  @type new_rapi_cert: bool
107
  @param new_rapi_cert: Whether to generate a new RAPI certificate
108
  @type new_spice_cert: bool
109
  @param new_spice_cert: Whether to generate a new SPICE certificate
110
  @type new_confd_hmac_key: bool
111
  @param new_confd_hmac_key: Whether to generate a new HMAC key
112
  @type new_cds: bool
113
  @param new_cds: Whether to generate a new cluster domain secret
114
  @type rapi_cert_pem: string
115
  @param rapi_cert_pem: New RAPI certificate in PEM format
116
  @type spice_cert_pem: string
117
  @param spice_cert_pem: New SPICE certificate in PEM format
118
  @type spice_cacert_pem: string
119
  @param spice_cacert_pem: Certificate of the CA that signed the SPICE
120
                           certificate, in PEM format
121
  @type cds: string
122
  @param cds: New cluster domain secret
123
  @type nodecert_file: string
124
  @param nodecert_file: optional override of the node cert file path
125
  @type rapicert_file: string
126
  @param rapicert_file: optional override of the rapi cert file path
127
  @type spicecert_file: string
128
  @param spicecert_file: optional override of the spice cert file path
129
  @type spicecacert_file: string
130
  @param spicecacert_file: optional override of the spice CA cert file path
131
  @type hmackey_file: string
132
  @param hmackey_file: optional override of the hmac key file path
133

134
  """
135
  # noded SSL certificate
136
  cluster_cert_exists = os.path.exists(nodecert_file)
137
  if new_cluster_cert or not cluster_cert_exists:
138
    if cluster_cert_exists:
139
      utils.CreateBackup(nodecert_file)
140

    
141
    logging.debug("Generating new cluster certificate at %s", nodecert_file)
142
    utils.GenerateSelfSignedSslCert(nodecert_file)
143

    
144
  # confd HMAC key
145
  if new_confd_hmac_key or not os.path.exists(hmackey_file):
146
    logging.debug("Writing new confd HMAC key to %s", hmackey_file)
147
    GenerateHmacKey(hmackey_file)
148

    
149
  # RAPI
150
  rapi_cert_exists = os.path.exists(rapicert_file)
151

    
152
  if rapi_cert_pem:
153
    # Assume rapi_pem contains a valid PEM-formatted certificate and key
154
    logging.debug("Writing RAPI certificate at %s", rapicert_file)
155
    utils.WriteFile(rapicert_file, data=rapi_cert_pem, backup=True)
156

    
157
  elif new_rapi_cert or not rapi_cert_exists:
158
    if rapi_cert_exists:
159
      utils.CreateBackup(rapicert_file)
160

    
161
    logging.debug("Generating new RAPI certificate at %s", rapicert_file)
162
    utils.GenerateSelfSignedSslCert(rapicert_file)
163

    
164
  # SPICE
165
  spice_cert_exists = os.path.exists(spicecert_file)
166
  spice_cacert_exists = os.path.exists(spicecacert_file)
167
  if spice_cert_pem:
168
    # spice_cert_pem implies also spice_cacert_pem
169
    logging.debug("Writing SPICE certificate at %s", spicecert_file)
170
    utils.WriteFile(spicecert_file, data=spice_cert_pem, backup=True)
171
    logging.debug("Writing SPICE CA certificate at %s", spicecacert_file)
172
    utils.WriteFile(spicecacert_file, data=spice_cacert_pem, backup=True)
173
  elif new_spice_cert or not spice_cert_exists:
174
    if spice_cert_exists:
175
      utils.CreateBackup(spicecert_file)
176
    if spice_cacert_exists:
177
      utils.CreateBackup(spicecacert_file)
178

    
179
    logging.debug("Generating new self-signed SPICE certificate at %s",
180
                  spicecert_file)
181
    (_, cert_pem) = utils.GenerateSelfSignedSslCert(spicecert_file)
182

    
183
    # Self-signed certificate -> the public certificate is also the CA public
184
    # certificate
185
    logging.debug("Writing the public certificate to %s",
186
                  spicecert_file)
187
    utils.io.WriteFile(spicecacert_file, mode=0400, data=cert_pem)
188

    
189
  # Cluster domain secret
190
  if cds:
191
    logging.debug("Writing cluster domain secret to %s", cds_file)
192
    utils.WriteFile(cds_file, data=cds, backup=True)
193

    
194
  elif new_cds or not os.path.exists(cds_file):
195
    logging.debug("Generating new cluster domain secret at %s", cds_file)
196
    GenerateHmacKey(cds_file)
197

    
198

    
199
def _InitGanetiServerSetup(master_name):
200
  """Setup the necessary configuration for the initial node daemon.
201

202
  This creates the nodepass file containing the shared password for
203
  the cluster, generates the SSL certificate and starts the node daemon.
204

205
  @type master_name: str
206
  @param master_name: Name of the master node
207

208
  """
209
  # Generate cluster secrets
210
  GenerateClusterCrypto(True, False, False, False, False)
211

    
212
  result = utils.RunCmd([pathutils.DAEMON_UTIL, "start", constants.NODED])
213
  if result.failed:
214
    raise errors.OpExecError("Could not start the node daemon, command %s"
215
                             " had exitcode %s and error %s" %
216
                             (result.cmd, result.exit_code, result.output))
217

    
218
  _WaitForNodeDaemon(master_name)
219

    
220

    
221
def _WaitForNodeDaemon(node_name):
222
  """Wait for node daemon to become responsive.
223

224
  """
225
  def _CheckNodeDaemon():
226
    # Pylint bug <http://www.logilab.org/ticket/35642>
227
    # pylint: disable=E1101
228
    result = rpc.BootstrapRunner().call_version([node_name])[node_name]
229
    if result.fail_msg:
230
      raise utils.RetryAgain()
231

    
232
  try:
233
    utils.Retry(_CheckNodeDaemon, 1.0, _DAEMON_READY_TIMEOUT)
234
  except utils.RetryTimeout:
235
    raise errors.OpExecError("Node daemon on %s didn't answer queries within"
236
                             " %s seconds" % (node_name, _DAEMON_READY_TIMEOUT))
237

    
238

    
239
def _WaitForMasterDaemon():
240
  """Wait for master daemon to become responsive.
241

242
  """
243
  def _CheckMasterDaemon():
244
    try:
245
      cl = luxi.Client()
246
      (cluster_name, ) = cl.QueryConfigValues(["cluster_name"])
247
    except Exception:
248
      raise utils.RetryAgain()
249

    
250
    logging.debug("Received cluster name %s from master", cluster_name)
251

    
252
  try:
253
    utils.Retry(_CheckMasterDaemon, 1.0, _DAEMON_READY_TIMEOUT)
254
  except utils.RetryTimeout:
255
    raise errors.OpExecError("Master daemon didn't answer queries within"
256
                             " %s seconds" % _DAEMON_READY_TIMEOUT)
257

    
258

    
259
def _WaitForSshDaemon(hostname, port, family):
260
  """Wait for SSH daemon to become responsive.
261

262
  """
263
  hostip = netutils.GetHostname(name=hostname, family=family).ip
264

    
265
  def _CheckSshDaemon():
266
    if netutils.TcpPing(hostip, port, timeout=1.0, live_port_needed=True):
267
      logging.debug("SSH daemon on %s:%s (IP address %s) has become"
268
                    " responsive", hostname, port, hostip)
269
    else:
270
      raise utils.RetryAgain()
271

    
272
  try:
273
    utils.Retry(_CheckSshDaemon, 1.0, _DAEMON_READY_TIMEOUT)
274
  except utils.RetryTimeout:
275
    raise errors.OpExecError("SSH daemon on %s:%s (IP address %s) didn't"
276
                             " become responsive within %s seconds" %
277
                             (hostname, port, hostip, _DAEMON_READY_TIMEOUT))
278

    
279

    
280
def RunNodeSetupCmd(cluster_name, node, basecmd, debug, verbose,
281
                    use_cluster_key, ask_key, strict_host_check, data):
282
  """Runs a command to configure something on a remote machine.
283

284
  @type cluster_name: string
285
  @param cluster_name: Cluster name
286
  @type node: string
287
  @param node: Node name
288
  @type basecmd: string
289
  @param basecmd: Base command (path on the remote machine)
290
  @type debug: bool
291
  @param debug: Enable debug output
292
  @type verbose: bool
293
  @param verbose: Enable verbose output
294
  @type use_cluster_key: bool
295
  @param use_cluster_key: See L{ssh.SshRunner.BuildCmd}
296
  @type ask_key: bool
297
  @param ask_key: See L{ssh.SshRunner.BuildCmd}
298
  @type strict_host_check: bool
299
  @param strict_host_check: See L{ssh.SshRunner.BuildCmd}
300
  @param data: JSON-serializable input data for script (passed to stdin)
301

302
  """
303
  cmd = [basecmd]
304

    
305
  # Pass --debug/--verbose to the external script if set on our invocation
306
  if debug:
307
    cmd.append("--debug")
308

    
309
  if verbose:
310
    cmd.append("--verbose")
311

    
312
  family = ssconf.SimpleStore().GetPrimaryIPFamily()
313
  srun = ssh.SshRunner(cluster_name,
314
                       ipv6=(family == netutils.IP6Address.family))
315
  scmd = srun.BuildCmd(node, constants.SSH_LOGIN_USER,
316
                       utils.ShellQuoteArgs(cmd),
317
                       batch=False, ask_key=ask_key, quiet=False,
318
                       strict_host_check=strict_host_check,
319
                       use_cluster_key=use_cluster_key)
320

    
321
  tempfh = tempfile.TemporaryFile()
322
  try:
323
    tempfh.write(serializer.DumpJson(data))
324
    tempfh.seek(0)
325

    
326
    result = utils.RunCmd(scmd, interactive=True, input_fd=tempfh)
327
  finally:
328
    tempfh.close()
329

    
330
  if result.failed:
331
    raise errors.OpExecError("Command '%s' failed: %s" %
332
                             (result.cmd, result.fail_reason))
333

    
334
  _WaitForSshDaemon(node, netutils.GetDaemonPort(constants.SSH), family)
335

    
336

    
337
def _PrepareFileStorage(enabled_disk_templates, file_storage_dir):
338
  """Checks if file storage is enabled and inits the dir.
339

340
  """
341
  if utils.storage.IsFileStorageEnabled(enabled_disk_templates):
342
    file_storage_dir = _InitFileStorageDir(file_storage_dir)
343
  else:
344
    file_storage_dir = ""
345
  return file_storage_dir
346

    
347

    
348
def _InitFileStorageDir(file_storage_dir):
349
  """Initialize if needed the file storage.
350

351
  @param file_storage_dir: the user-supplied value
352
  @return: either empty string (if file storage was disabled at build
353
      time) or the normalized path to the storage directory
354

355
  """
356
  file_storage_dir = os.path.normpath(file_storage_dir)
357

    
358
  if not os.path.isabs(file_storage_dir):
359
    raise errors.OpPrereqError("File storage directory '%s' is not an absolute"
360
                               " path" % file_storage_dir, errors.ECODE_INVAL)
361

    
362
  if not os.path.exists(file_storage_dir):
363
    try:
364
      os.makedirs(file_storage_dir, 0750)
365
    except OSError, err:
366
      raise errors.OpPrereqError("Cannot create file storage directory"
367
                                 " '%s': %s" % (file_storage_dir, err),
368
                                 errors.ECODE_ENVIRON)
369

    
370
  if not os.path.isdir(file_storage_dir):
371
    raise errors.OpPrereqError("The file storage directory '%s' is not"
372
                               " a directory." % file_storage_dir,
373
                               errors.ECODE_ENVIRON)
374

    
375
  # FIXME: check here if the file_storage_dir is in the set of allowed dirs
376
  return file_storage_dir
377

    
378

    
379
def _InitCheckEnabledDiskTemplates(enabled_disk_templates):
380
  """Checks the sanity of the enabled disk templates.
381

382
  """
383
  if not enabled_disk_templates:
384
    raise errors.OpPrereqError("Enabled disk templates list must contain at"
385
                               " least one member", errors.ECODE_INVAL)
386
  invalid_disk_templates = \
387
    set(enabled_disk_templates) - constants.DISK_TEMPLATES
388
  if invalid_disk_templates:
389
    raise errors.OpPrereqError("Enabled disk templates list contains invalid"
390
                               " entries: %s" % invalid_disk_templates,
391
                               errors.ECODE_INVAL)
392

    
393

    
394
def InitCluster(cluster_name, mac_prefix, # pylint: disable=R0913, R0914
395
                master_netmask, master_netdev, file_storage_dir,
396
                shared_file_storage_dir, candidate_pool_size, secondary_ip=None,
397
                vg_name=None, beparams=None, nicparams=None, ndparams=None,
398
                hvparams=None, diskparams=None, enabled_hypervisors=None,
399
                modify_etc_hosts=True, modify_ssh_setup=True,
400
                maintain_node_health=False, drbd_helper=None, uid_pool=None,
401
                default_iallocator=None, primary_ip_version=None, ipolicy=None,
402
                prealloc_wipe_disks=False, use_external_mip_script=False,
403
                hv_state=None, disk_state=None, enabled_disk_templates=None):
404
  """Initialise the cluster.
405

406
  @type candidate_pool_size: int
407
  @param candidate_pool_size: master candidate pool size
408
  @type enabled_disk_templates: list of string
409
  @param enabled_disk_templates: list of disk_templates to be used in this
410
    cluster
411

412
  """
413
  # TODO: complete the docstring
414
  if config.ConfigWriter.IsCluster():
415
    raise errors.OpPrereqError("Cluster is already initialised",
416
                               errors.ECODE_STATE)
417

    
418
  if not enabled_hypervisors:
419
    raise errors.OpPrereqError("Enabled hypervisors list must contain at"
420
                               " least one member", errors.ECODE_INVAL)
421
  invalid_hvs = set(enabled_hypervisors) - constants.HYPER_TYPES
422
  if invalid_hvs:
423
    raise errors.OpPrereqError("Enabled hypervisors contains invalid"
424
                               " entries: %s" % invalid_hvs,
425
                               errors.ECODE_INVAL)
426

    
427
  _InitCheckEnabledDiskTemplates(enabled_disk_templates)
428

    
429
  try:
430
    ipcls = netutils.IPAddress.GetClassFromIpVersion(primary_ip_version)
431
  except errors.ProgrammerError:
432
    raise errors.OpPrereqError("Invalid primary ip version: %d." %
433
                               primary_ip_version, errors.ECODE_INVAL)
434

    
435
  hostname = netutils.GetHostname(family=ipcls.family)
436
  if not ipcls.IsValid(hostname.ip):
437
    raise errors.OpPrereqError("This host's IP (%s) is not a valid IPv%d"
438
                               " address." % (hostname.ip, primary_ip_version),
439
                               errors.ECODE_INVAL)
440

    
441
  if ipcls.IsLoopback(hostname.ip):
442
    raise errors.OpPrereqError("This host's IP (%s) resolves to a loopback"
443
                               " address. Please fix DNS or %s." %
444
                               (hostname.ip, pathutils.ETC_HOSTS),
445
                               errors.ECODE_ENVIRON)
446

    
447
  if not ipcls.Own(hostname.ip):
448
    raise errors.OpPrereqError("Inconsistency: this host's name resolves"
449
                               " to %s,\nbut this ip address does not"
450
                               " belong to this host" %
451
                               hostname.ip, errors.ECODE_ENVIRON)
452

    
453
  clustername = netutils.GetHostname(name=cluster_name, family=ipcls.family)
454

    
455
  if netutils.TcpPing(clustername.ip, constants.DEFAULT_NODED_PORT, timeout=5):
456
    raise errors.OpPrereqError("Cluster IP already active",
457
                               errors.ECODE_NOTUNIQUE)
458

    
459
  if not secondary_ip:
460
    if primary_ip_version == constants.IP6_VERSION:
461
      raise errors.OpPrereqError("When using a IPv6 primary address, a valid"
462
                                 " IPv4 address must be given as secondary",
463
                                 errors.ECODE_INVAL)
464
    secondary_ip = hostname.ip
465

    
466
  if not netutils.IP4Address.IsValid(secondary_ip):
467
    raise errors.OpPrereqError("Secondary IP address (%s) has to be a valid"
468
                               " IPv4 address." % secondary_ip,
469
                               errors.ECODE_INVAL)
470

    
471
  if not netutils.IP4Address.Own(secondary_ip):
472
    raise errors.OpPrereqError("You gave %s as secondary IP,"
473
                               " but it does not belong to this host." %
474
                               secondary_ip, errors.ECODE_ENVIRON)
475

    
476
  if master_netmask is not None:
477
    if not ipcls.ValidateNetmask(master_netmask):
478
      raise errors.OpPrereqError("CIDR netmask (%s) not valid for IPv%s " %
479
                                  (master_netmask, primary_ip_version),
480
                                 errors.ECODE_INVAL)
481
  else:
482
    master_netmask = ipcls.iplen
483

    
484
  if vg_name:
485
    # Check if volume group is valid
486
    vgstatus = utils.CheckVolumeGroupSize(utils.ListVolumeGroups(), vg_name,
487
                                          constants.MIN_VG_SIZE)
488
    if vgstatus:
489
      raise errors.OpPrereqError("Error: %s" % vgstatus, errors.ECODE_INVAL)
490

    
491
  if drbd_helper is not None:
492
    try:
493
      curr_helper = drbd.DRBD8.GetUsermodeHelper()
494
    except errors.BlockDeviceError, err:
495
      raise errors.OpPrereqError("Error while checking drbd helper"
496
                                 " (specify --no-drbd-storage if you are not"
497
                                 " using drbd): %s" % str(err),
498
                                 errors.ECODE_ENVIRON)
499
    if drbd_helper != curr_helper:
500
      raise errors.OpPrereqError("Error: requiring %s as drbd helper but %s"
501
                                 " is the current helper" % (drbd_helper,
502
                                                             curr_helper),
503
                                 errors.ECODE_INVAL)
504

    
505
  logging.debug("Stopping daemons (if any are running)")
506
  result = utils.RunCmd([pathutils.DAEMON_UTIL, "stop-all"])
507
  if result.failed:
508
    raise errors.OpExecError("Could not stop daemons, command %s"
509
                             " had exitcode %s and error '%s'" %
510
                             (result.cmd, result.exit_code, result.output))
511

    
512
  file_storage_dir = _PrepareFileStorage(enabled_disk_templates,
513
                                         file_storage_dir)
514

    
515
  if constants.ENABLE_SHARED_FILE_STORAGE:
516
    shared_file_storage_dir = _InitFileStorageDir(shared_file_storage_dir)
517
  else:
518
    shared_file_storage_dir = ""
519

    
520
  if not re.match("^[0-9a-z]{2}:[0-9a-z]{2}:[0-9a-z]{2}$", mac_prefix):
521
    raise errors.OpPrereqError("Invalid mac prefix given '%s'" % mac_prefix,
522
                               errors.ECODE_INVAL)
523

    
524
  result = utils.RunCmd(["ip", "link", "show", "dev", master_netdev])
525
  if result.failed:
526
    raise errors.OpPrereqError("Invalid master netdev given (%s): '%s'" %
527
                               (master_netdev,
528
                                result.output.strip()), errors.ECODE_INVAL)
529

    
530
  dirs = [(pathutils.RUN_DIR, constants.RUN_DIRS_MODE)]
531
  utils.EnsureDirs(dirs)
532

    
533
  objects.UpgradeBeParams(beparams)
534
  utils.ForceDictType(beparams, constants.BES_PARAMETER_TYPES)
535
  utils.ForceDictType(nicparams, constants.NICS_PARAMETER_TYPES)
536

    
537
  objects.NIC.CheckParameterSyntax(nicparams)
538

    
539
  full_ipolicy = objects.FillIPolicy(constants.IPOLICY_DEFAULTS, ipolicy)
540

    
541
  if ndparams is not None:
542
    utils.ForceDictType(ndparams, constants.NDS_PARAMETER_TYPES)
543
  else:
544
    ndparams = dict(constants.NDC_DEFAULTS)
545

    
546
  # This is ugly, as we modify the dict itself
547
  # FIXME: Make utils.ForceDictType pure functional or write a wrapper
548
  # around it
549
  if hv_state:
550
    for hvname, hvs_data in hv_state.items():
551
      utils.ForceDictType(hvs_data, constants.HVSTS_PARAMETER_TYPES)
552
      hv_state[hvname] = objects.Cluster.SimpleFillHvState(hvs_data)
553
  else:
554
    hv_state = dict((hvname, constants.HVST_DEFAULTS)
555
                    for hvname in enabled_hypervisors)
556

    
557
  # FIXME: disk_state has no default values yet
558
  if disk_state:
559
    for storage, ds_data in disk_state.items():
560
      if storage not in constants.DS_VALID_TYPES:
561
        raise errors.OpPrereqError("Invalid storage type in disk state: %s" %
562
                                   storage, errors.ECODE_INVAL)
563
      for ds_name, state in ds_data.items():
564
        utils.ForceDictType(state, constants.DSS_PARAMETER_TYPES)
565
        ds_data[ds_name] = objects.Cluster.SimpleFillDiskState(state)
566

    
567
  # hvparams is a mapping of hypervisor->hvparams dict
568
  for hv_name, hv_params in hvparams.iteritems():
569
    utils.ForceDictType(hv_params, constants.HVS_PARAMETER_TYPES)
570
    hv_class = hypervisor.GetHypervisor(hv_name)
571
    hv_class.CheckParameterSyntax(hv_params)
572

    
573
  # diskparams is a mapping of disk-template->diskparams dict
574
  for template, dt_params in diskparams.items():
575
    param_keys = set(dt_params.keys())
576
    default_param_keys = set(constants.DISK_DT_DEFAULTS[template].keys())
577
    if not (param_keys <= default_param_keys):
578
      unknown_params = param_keys - default_param_keys
579
      raise errors.OpPrereqError("Invalid parameters for disk template %s:"
580
                                 " %s" % (template,
581
                                          utils.CommaJoin(unknown_params)),
582
                                 errors.ECODE_INVAL)
583
    utils.ForceDictType(dt_params, constants.DISK_DT_TYPES)
584
    if template == constants.DT_DRBD8 and vg_name is not None:
585
      # The default METAVG value is equal to the VG name set at init time,
586
      # if provided
587
      dt_params[constants.DRBD_DEFAULT_METAVG] = vg_name
588

    
589
  try:
590
    utils.VerifyDictOptions(diskparams, constants.DISK_DT_DEFAULTS)
591
  except errors.OpPrereqError, err:
592
    raise errors.OpPrereqError("While verify diskparam options: %s" % err,
593
                               errors.ECODE_INVAL)
594

    
595
  # set up ssh config and /etc/hosts
596
  sshline = utils.ReadFile(pathutils.SSH_HOST_RSA_PUB)
597
  sshkey = sshline.split(" ")[1]
598

    
599
  if modify_etc_hosts:
600
    utils.AddHostToEtcHosts(hostname.name, hostname.ip)
601

    
602
  if modify_ssh_setup:
603
    _InitSSHSetup()
604

    
605
  if default_iallocator is not None:
606
    alloc_script = utils.FindFile(default_iallocator,
607
                                  constants.IALLOCATOR_SEARCH_PATH,
608
                                  os.path.isfile)
609
    if alloc_script is None:
610
      raise errors.OpPrereqError("Invalid default iallocator script '%s'"
611
                                 " specified" % default_iallocator,
612
                                 errors.ECODE_INVAL)
613
  elif constants.HTOOLS:
614
    # htools was enabled at build-time, we default to it
615
    if utils.FindFile(constants.IALLOC_HAIL,
616
                      constants.IALLOCATOR_SEARCH_PATH,
617
                      os.path.isfile):
618
      default_iallocator = constants.IALLOC_HAIL
619

    
620
  now = time.time()
621

    
622
  # init of cluster config file
623
  cluster_config = objects.Cluster(
624
    serial_no=1,
625
    rsahostkeypub=sshkey,
626
    highest_used_port=(constants.FIRST_DRBD_PORT - 1),
627
    mac_prefix=mac_prefix,
628
    volume_group_name=vg_name,
629
    tcpudp_port_pool=set(),
630
    master_ip=clustername.ip,
631
    master_netmask=master_netmask,
632
    master_netdev=master_netdev,
633
    cluster_name=clustername.name,
634
    file_storage_dir=file_storage_dir,
635
    shared_file_storage_dir=shared_file_storage_dir,
636
    enabled_hypervisors=enabled_hypervisors,
637
    beparams={constants.PP_DEFAULT: beparams},
638
    nicparams={constants.PP_DEFAULT: nicparams},
639
    ndparams=ndparams,
640
    hvparams=hvparams,
641
    diskparams=diskparams,
642
    candidate_pool_size=candidate_pool_size,
643
    modify_etc_hosts=modify_etc_hosts,
644
    modify_ssh_setup=modify_ssh_setup,
645
    uid_pool=uid_pool,
646
    ctime=now,
647
    mtime=now,
648
    maintain_node_health=maintain_node_health,
649
    drbd_usermode_helper=drbd_helper,
650
    default_iallocator=default_iallocator,
651
    primary_ip_family=ipcls.family,
652
    prealloc_wipe_disks=prealloc_wipe_disks,
653
    use_external_mip_script=use_external_mip_script,
654
    ipolicy=full_ipolicy,
655
    hv_state_static=hv_state,
656
    disk_state_static=disk_state,
657
    enabled_disk_templates=enabled_disk_templates,
658
    )
659
  master_node_config = objects.Node(name=hostname.name,
660
                                    primary_ip=hostname.ip,
661
                                    secondary_ip=secondary_ip,
662
                                    serial_no=1,
663
                                    master_candidate=True,
664
                                    offline=False, drained=False,
665
                                    ctime=now, mtime=now,
666
                                    )
667
  InitConfig(constants.CONFIG_VERSION, cluster_config, master_node_config)
668
  cfg = config.ConfigWriter(offline=True)
669
  ssh.WriteKnownHostsFile(cfg, pathutils.SSH_KNOWN_HOSTS_FILE)
670
  cfg.Update(cfg.GetClusterInfo(), logging.error)
671
  ssconf.WriteSsconfFiles(cfg.GetSsconfValues())
672

    
673
  # set up the inter-node password and certificate
674
  _InitGanetiServerSetup(hostname.name)
675

    
676
  logging.debug("Starting daemons")
677
  result = utils.RunCmd([pathutils.DAEMON_UTIL, "start-all"])
678
  if result.failed:
679
    raise errors.OpExecError("Could not start daemons, command %s"
680
                             " had exitcode %s and error %s" %
681
                             (result.cmd, result.exit_code, result.output))
682

    
683
  _WaitForMasterDaemon()
684

    
685

    
686
def InitConfig(version, cluster_config, master_node_config,
687
               cfg_file=pathutils.CLUSTER_CONF_FILE):
688
  """Create the initial cluster configuration.
689

690
  It will contain the current node, which will also be the master
691
  node, and no instances.
692

693
  @type version: int
694
  @param version: configuration version
695
  @type cluster_config: L{objects.Cluster}
696
  @param cluster_config: cluster configuration
697
  @type master_node_config: L{objects.Node}
698
  @param master_node_config: master node configuration
699
  @type cfg_file: string
700
  @param cfg_file: configuration file path
701

702
  """
703
  uuid_generator = config.TemporaryReservationManager()
704
  cluster_config.uuid = uuid_generator.Generate([], utils.NewUUID,
705
                                                _INITCONF_ECID)
706
  master_node_config.uuid = uuid_generator.Generate([], utils.NewUUID,
707
                                                    _INITCONF_ECID)
708
  cluster_config.master_node = master_node_config.uuid
709
  nodes = {
710
    master_node_config.uuid: master_node_config,
711
    }
712
  default_nodegroup = objects.NodeGroup(
713
    uuid=uuid_generator.Generate([], utils.NewUUID, _INITCONF_ECID),
714
    name=constants.INITIAL_NODE_GROUP_NAME,
715
    members=[master_node_config.uuid],
716
    diskparams={},
717
    )
718
  nodegroups = {
719
    default_nodegroup.uuid: default_nodegroup,
720
    }
721
  now = time.time()
722
  config_data = objects.ConfigData(version=version,
723
                                   cluster=cluster_config,
724
                                   nodegroups=nodegroups,
725
                                   nodes=nodes,
726
                                   instances={},
727
                                   networks={},
728
                                   serial_no=1,
729
                                   ctime=now, mtime=now)
730
  utils.WriteFile(cfg_file,
731
                  data=serializer.Dump(config_data.ToDict()),
732
                  mode=0600)
733

    
734

    
735
def FinalizeClusterDestroy(master_uuid):
736
  """Execute the last steps of cluster destroy
737

738
  This function shuts down all the daemons, completing the destroy
739
  begun in cmdlib.LUDestroyOpcode.
740

741
  """
742
  cfg = config.ConfigWriter()
743
  modify_ssh_setup = cfg.GetClusterInfo().modify_ssh_setup
744
  runner = rpc.BootstrapRunner()
745

    
746
  master_name = cfg.GetNodeName(master_uuid)
747

    
748
  master_params = cfg.GetMasterNetworkParameters()
749
  master_params.uuid = master_uuid
750
  ems = cfg.GetUseExternalMipScript()
751
  result = runner.call_node_deactivate_master_ip(master_name, master_params,
752
                                                 ems)
753

    
754
  msg = result.fail_msg
755
  if msg:
756
    logging.warning("Could not disable the master IP: %s", msg)
757

    
758
  result = runner.call_node_stop_master(master_name)
759
  msg = result.fail_msg
760
  if msg:
761
    logging.warning("Could not disable the master role: %s", msg)
762

    
763
  result = runner.call_node_leave_cluster(master_name, modify_ssh_setup)
764
  msg = result.fail_msg
765
  if msg:
766
    logging.warning("Could not shutdown the node daemon and cleanup"
767
                    " the node: %s", msg)
768

    
769

    
770
def SetupNodeDaemon(opts, cluster_name, node):
771
  """Add a node to the cluster.
772

773
  This function must be called before the actual opcode, and will ssh
774
  to the remote node, copy the needed files, and start ganeti-noded,
775
  allowing the master to do the rest via normal rpc calls.
776

777
  @param cluster_name: the cluster name
778
  @param node: the name of the new node
779

780
  """
781
  data = {
782
    constants.NDS_CLUSTER_NAME: cluster_name,
783
    constants.NDS_NODE_DAEMON_CERTIFICATE:
784
      utils.ReadFile(pathutils.NODED_CERT_FILE),
785
    constants.NDS_SSCONF: ssconf.SimpleStore().ReadAll(),
786
    constants.NDS_START_NODE_DAEMON: True,
787
    }
788

    
789
  RunNodeSetupCmd(cluster_name, node, pathutils.NODE_DAEMON_SETUP,
790
                  opts.debug, opts.verbose,
791
                  True, opts.ssh_key_check, opts.ssh_key_check, data)
792

    
793
  _WaitForNodeDaemon(node)
794

    
795

    
796
def MasterFailover(no_voting=False):
797
  """Failover the master node.
798

799
  This checks that we are not already the master, and will cause the
800
  current master to cease being master, and the non-master to become
801
  new master.
802

803
  @type no_voting: boolean
804
  @param no_voting: force the operation without remote nodes agreement
805
                      (dangerous)
806

807
  """
808
  sstore = ssconf.SimpleStore()
809

    
810
  old_master, new_master = ssconf.GetMasterAndMyself(sstore)
811
  node_names = sstore.GetNodeList()
812
  mc_list = sstore.GetMasterCandidates()
813

    
814
  if old_master == new_master:
815
    raise errors.OpPrereqError("This commands must be run on the node"
816
                               " where you want the new master to be."
817
                               " %s is already the master" %
818
                               old_master, errors.ECODE_INVAL)
819

    
820
  if new_master not in mc_list:
821
    mc_no_master = [name for name in mc_list if name != old_master]
822
    raise errors.OpPrereqError("This node is not among the nodes marked"
823
                               " as master candidates. Only these nodes"
824
                               " can become masters. Current list of"
825
                               " master candidates is:\n"
826
                               "%s" % ("\n".join(mc_no_master)),
827
                               errors.ECODE_STATE)
828

    
829
  if not no_voting:
830
    vote_list = GatherMasterVotes(node_names)
831

    
832
    if vote_list:
833
      voted_master = vote_list[0][0]
834
      if voted_master is None:
835
        raise errors.OpPrereqError("Cluster is inconsistent, most nodes did"
836
                                   " not respond.", errors.ECODE_ENVIRON)
837
      elif voted_master != old_master:
838
        raise errors.OpPrereqError("I have a wrong configuration, I believe"
839
                                   " the master is %s but the other nodes"
840
                                   " voted %s. Please resync the configuration"
841
                                   " of this node." %
842
                                   (old_master, voted_master),
843
                                   errors.ECODE_STATE)
844
  # end checks
845

    
846
  rcode = 0
847

    
848
  logging.info("Setting master to %s, old master: %s", new_master, old_master)
849

    
850
  try:
851
    # instantiate a real config writer, as we now know we have the
852
    # configuration data
853
    cfg = config.ConfigWriter(accept_foreign=True)
854

    
855
    old_master_node = cfg.GetNodeInfoByName(old_master)
856
    if old_master_node is None:
857
      raise errors.OpPrereqError("Could not find old master node '%s' in"
858
                                 " cluster configuration." % old_master,
859
                                 errors.ECODE_NOENT)
860

    
861
    cluster_info = cfg.GetClusterInfo()
862
    new_master_node = cfg.GetNodeInfoByName(new_master)
863
    if new_master_node is None:
864
      raise errors.OpPrereqError("Could not find new master node '%s' in"
865
                                 " cluster configuration." % new_master,
866
                                 errors.ECODE_NOENT)
867

    
868
    cluster_info.master_node = new_master_node.uuid
869
    # this will also regenerate the ssconf files, since we updated the
870
    # cluster info
871
    cfg.Update(cluster_info, logging.error)
872
  except errors.ConfigurationError, err:
873
    logging.error("Error while trying to set the new master: %s",
874
                  str(err))
875
    return 1
876

    
877
  # if cfg.Update worked, then it means the old master daemon won't be
878
  # able now to write its own config file (we rely on locking in both
879
  # backend.UploadFile() and ConfigWriter._Write(); hence the next
880
  # step is to kill the old master
881

    
882
  logging.info("Stopping the master daemon on node %s", old_master)
883

    
884
  runner = rpc.BootstrapRunner()
885
  master_params = cfg.GetMasterNetworkParameters()
886
  master_params.uuid = old_master_node.uuid
887
  ems = cfg.GetUseExternalMipScript()
888
  result = runner.call_node_deactivate_master_ip(old_master,
889
                                                 master_params, ems)
890

    
891
  msg = result.fail_msg
892
  if msg:
893
    logging.warning("Could not disable the master IP: %s", msg)
894

    
895
  result = runner.call_node_stop_master(old_master)
896
  msg = result.fail_msg
897
  if msg:
898
    logging.error("Could not disable the master role on the old master"
899
                  " %s, please disable manually: %s", old_master, msg)
900

    
901
  logging.info("Checking master IP non-reachability...")
902

    
903
  master_ip = sstore.GetMasterIP()
904
  total_timeout = 30
905

    
906
  # Here we have a phase where no master should be running
907
  def _check_ip():
908
    if netutils.TcpPing(master_ip, constants.DEFAULT_NODED_PORT):
909
      raise utils.RetryAgain()
910

    
911
  try:
912
    utils.Retry(_check_ip, (1, 1.5, 5), total_timeout)
913
  except utils.RetryTimeout:
914
    logging.warning("The master IP is still reachable after %s seconds,"
915
                    " continuing but activating the master on the current"
916
                    " node will probably fail", total_timeout)
917

    
918
  if jstore.CheckDrainFlag():
919
    logging.info("Undraining job queue")
920
    jstore.SetDrainFlag(False)
921

    
922
  logging.info("Starting the master daemons on the new master")
923

    
924
  result = rpc.BootstrapRunner().call_node_start_master_daemons(new_master,
925
                                                                no_voting)
926
  msg = result.fail_msg
927
  if msg:
928
    logging.error("Could not start the master role on the new master"
929
                  " %s, please check: %s", new_master, msg)
930
    rcode = 1
931

    
932
  logging.info("Master failed over from %s to %s", old_master, new_master)
933
  return rcode
934

    
935

    
936
def GetMaster():
937
  """Returns the current master node.
938

939
  This is a separate function in bootstrap since it's needed by
940
  gnt-cluster, and instead of importing directly ssconf, it's better
941
  to abstract it in bootstrap, where we do use ssconf in other
942
  functions too.
943

944
  """
945
  sstore = ssconf.SimpleStore()
946

    
947
  old_master, _ = ssconf.GetMasterAndMyself(sstore)
948

    
949
  return old_master
950

    
951

    
952
def GatherMasterVotes(node_names):
953
  """Check the agreement on who is the master.
954

955
  This function will return a list of (node, number of votes), ordered
956
  by the number of votes. Errors will be denoted by the key 'None'.
957

958
  Note that the sum of votes is the number of nodes this machine
959
  knows, whereas the number of entries in the list could be different
960
  (if some nodes vote for another master).
961

962
  We remove ourselves from the list since we know that (bugs aside)
963
  since we use the same source for configuration information for both
964
  backend and boostrap, we'll always vote for ourselves.
965

966
  @type node_names: list
967
  @param node_names: the list of nodes to query for master info; the current
968
      node will be removed if it is in the list
969
  @rtype: list
970
  @return: list of (node, votes)
971

972
  """
973
  myself = netutils.Hostname.GetSysName()
974
  try:
975
    node_names.remove(myself)
976
  except ValueError:
977
    pass
978
  if not node_names:
979
    # no nodes left (eventually after removing myself)
980
    return []
981
  results = rpc.BootstrapRunner().call_master_info(node_names)
982
  if not isinstance(results, dict):
983
    # this should not happen (unless internal error in rpc)
984
    logging.critical("Can't complete rpc call, aborting master startup")
985
    return [(None, len(node_names))]
986
  votes = {}
987
  for node_name in results:
988
    nres = results[node_name]
989
    data = nres.payload
990
    msg = nres.fail_msg
991
    fail = False
992
    if msg:
993
      logging.warning("Error contacting node %s: %s", node_name, msg)
994
      fail = True
995
    # for now we accept both length 3, 4 and 5 (data[3] is primary ip version
996
    # and data[4] is the master netmask)
997
    elif not isinstance(data, (tuple, list)) or len(data) < 3:
998
      logging.warning("Invalid data received from node %s: %s",
999
                      node_name, data)
1000
      fail = True
1001
    if fail:
1002
      if None not in votes:
1003
        votes[None] = 0
1004
      votes[None] += 1
1005
      continue
1006
    master_node = data[2]
1007
    if master_node not in votes:
1008
      votes[master_node] = 0
1009
    votes[master_node] += 1
1010

    
1011
  vote_list = [v for v in votes.items()]
1012
  # sort first on number of votes then on name, since we want None
1013
  # sorted later if we have the half of the nodes not responding, and
1014
  # half voting all for the same master
1015
  vote_list.sort(key=lambda x: (x[1], x[0]), reverse=True)
1016

    
1017
  return vote_list