Statistics
| Branch: | Tag: | Revision:

root / lib / bootstrap.py @ 7c74bbe0

History | View | Annotate | Download (30.6 kB)

1
#
2
#
3

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

    
21

    
22
"""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

    
32
from ganeti import rpc
33
from ganeti import ssh
34
from ganeti import utils
35
from ganeti import errors
36
from ganeti import config
37
from ganeti import constants
38
from ganeti import objects
39
from ganeti import ssconf
40
from ganeti import serializer
41
from ganeti import hypervisor
42
from ganeti import bdev
43
from ganeti import netutils
44
from ganeti import backend
45
from ganeti import luxi
46
from ganeti import jstore
47

    
48

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

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

    
55

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

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

62
  """
63
  priv_key, pub_key, auth_keys = ssh.GetUserFiles(constants.GANETI_RUNAS)
64

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

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

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

    
79

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

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

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

    
90

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

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

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

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

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

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

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

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

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

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

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

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

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

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

    
197

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

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

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

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

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

    
217
  _WaitForNodeDaemon(master_name)
218

    
219

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

223
  """
224
  def _CheckNodeDaemon():
225
    result = rpc.BootstrapRunner().call_version([node_name])[node_name]
226
    if result.fail_msg:
227
      raise utils.RetryAgain()
228

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

    
235

    
236
def _WaitForMasterDaemon():
237
  """Wait for master daemon to become responsive.
238

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

    
247
    logging.debug("Received cluster name %s from master", cluster_name)
248

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

    
255

    
256
def _InitFileStorage(file_storage_dir):
257
  """Initialize if needed the file storage.
258

259
  @param file_storage_dir: the user-supplied value
260
  @return: either empty string (if file storage was disabled at build
261
      time) or the normalized path to the storage directory
262

263
  """
264
  file_storage_dir = os.path.normpath(file_storage_dir)
265

    
266
  if not os.path.isabs(file_storage_dir):
267
    raise errors.OpPrereqError("File storage directory '%s' is not an absolute"
268
                               " path" % file_storage_dir, errors.ECODE_INVAL)
269

    
270
  if not os.path.exists(file_storage_dir):
271
    try:
272
      os.makedirs(file_storage_dir, 0750)
273
    except OSError, err:
274
      raise errors.OpPrereqError("Cannot create file storage directory"
275
                                 " '%s': %s" % (file_storage_dir, err),
276
                                 errors.ECODE_ENVIRON)
277

    
278
  if not os.path.isdir(file_storage_dir):
279
    raise errors.OpPrereqError("The file storage directory '%s' is not"
280
                               " a directory." % file_storage_dir,
281
                               errors.ECODE_ENVIRON)
282
  return file_storage_dir
283

    
284

    
285
def InitCluster(cluster_name, mac_prefix, # pylint: disable=R0913
286
                master_netmask, master_netdev, file_storage_dir,
287
                shared_file_storage_dir, candidate_pool_size, secondary_ip=None,
288
                vg_name=None, beparams=None, nicparams=None, ndparams=None,
289
                hvparams=None, enabled_hypervisors=None, modify_etc_hosts=True,
290
                modify_ssh_setup=True, maintain_node_health=False,
291
                drbd_helper=None, uid_pool=None, default_iallocator=None,
292
                primary_ip_version=None, prealloc_wipe_disks=False):
293
  """Initialise the cluster.
294

295
  @type candidate_pool_size: int
296
  @param candidate_pool_size: master candidate pool size
297

298
  """
299
  # TODO: complete the docstring
300
  if config.ConfigWriter.IsCluster():
301
    raise errors.OpPrereqError("Cluster is already initialised",
302
                               errors.ECODE_STATE)
303

    
304
  if not enabled_hypervisors:
305
    raise errors.OpPrereqError("Enabled hypervisors list must contain at"
306
                               " least one member", errors.ECODE_INVAL)
307
  invalid_hvs = set(enabled_hypervisors) - constants.HYPER_TYPES
308
  if invalid_hvs:
309
    raise errors.OpPrereqError("Enabled hypervisors contains invalid"
310
                               " entries: %s" % invalid_hvs,
311
                               errors.ECODE_INVAL)
312

    
313
  try:
314
    ipcls = netutils.IPAddress.GetClassFromIpVersion(primary_ip_version)
315
  except errors.ProgrammerError:
316
    raise errors.OpPrereqError("Invalid primary ip version: %d." %
317
                               primary_ip_version)
318

    
319
  hostname = netutils.GetHostname(family=ipcls.family)
320
  if not ipcls.IsValid(hostname.ip):
321
    raise errors.OpPrereqError("This host's IP (%s) is not a valid IPv%d"
322
                               " address." % (hostname.ip, primary_ip_version))
323

    
324
  if ipcls.IsLoopback(hostname.ip):
325
    raise errors.OpPrereqError("This host's IP (%s) resolves to a loopback"
326
                               " address. Please fix DNS or %s." %
327
                               (hostname.ip, constants.ETC_HOSTS),
328
                               errors.ECODE_ENVIRON)
329

    
330
  if not ipcls.Own(hostname.ip):
331
    raise errors.OpPrereqError("Inconsistency: this host's name resolves"
332
                               " to %s,\nbut this ip address does not"
333
                               " belong to this host" %
334
                               hostname.ip, errors.ECODE_ENVIRON)
335

    
336
  clustername = netutils.GetHostname(name=cluster_name, family=ipcls.family)
337

    
338
  if netutils.TcpPing(clustername.ip, constants.DEFAULT_NODED_PORT, timeout=5):
339
    raise errors.OpPrereqError("Cluster IP already active",
340
                               errors.ECODE_NOTUNIQUE)
341

    
342
  if not secondary_ip:
343
    if primary_ip_version == constants.IP6_VERSION:
344
      raise errors.OpPrereqError("When using a IPv6 primary address, a valid"
345
                                 " IPv4 address must be given as secondary",
346
                                 errors.ECODE_INVAL)
347
    secondary_ip = hostname.ip
348

    
349
  if not netutils.IP4Address.IsValid(secondary_ip):
350
    raise errors.OpPrereqError("Secondary IP address (%s) has to be a valid"
351
                               " IPv4 address." % secondary_ip,
352
                               errors.ECODE_INVAL)
353

    
354
  if not netutils.IP4Address.Own(secondary_ip):
355
    raise errors.OpPrereqError("You gave %s as secondary IP,"
356
                               " but it does not belong to this host." %
357
                               secondary_ip, errors.ECODE_ENVIRON)
358

    
359
  if master_netmask is not None:
360
    if not ipcls.ValidateNetmask(master_netmask):
361
      raise errors.OpPrereqError("CIDR netmask (%s) not valid for IPv%s " %
362
                                  (master_netmask, primary_ip_version))
363
  else:
364
    master_netmask = ipcls.iplen
365

    
366
  if vg_name is not None:
367
    # Check if volume group is valid
368
    vgstatus = utils.CheckVolumeGroupSize(utils.ListVolumeGroups(), vg_name,
369
                                          constants.MIN_VG_SIZE)
370
    if vgstatus:
371
      raise errors.OpPrereqError("Error: %s\nspecify --no-lvm-storage if"
372
                                 " you are not using lvm" % vgstatus,
373
                                 errors.ECODE_INVAL)
374

    
375
  if drbd_helper is not None:
376
    try:
377
      curr_helper = bdev.BaseDRBD.GetUsermodeHelper()
378
    except errors.BlockDeviceError, err:
379
      raise errors.OpPrereqError("Error while checking drbd helper"
380
                                 " (specify --no-drbd-storage if you are not"
381
                                 " using drbd): %s" % str(err),
382
                                 errors.ECODE_ENVIRON)
383
    if drbd_helper != curr_helper:
384
      raise errors.OpPrereqError("Error: requiring %s as drbd helper but %s"
385
                                 " is the current helper" % (drbd_helper,
386
                                                             curr_helper),
387
                                 errors.ECODE_INVAL)
388

    
389
  if constants.ENABLE_FILE_STORAGE:
390
    file_storage_dir = _InitFileStorage(file_storage_dir)
391
  else:
392
    file_storage_dir = ""
393

    
394
  if constants.ENABLE_SHARED_FILE_STORAGE:
395
    shared_file_storage_dir = _InitFileStorage(shared_file_storage_dir)
396
  else:
397
    shared_file_storage_dir = ""
398

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

    
403
  result = utils.RunCmd(["ip", "link", "show", "dev", master_netdev])
404
  if result.failed:
405
    raise errors.OpPrereqError("Invalid master netdev given (%s): '%s'" %
406
                               (master_netdev,
407
                                result.output.strip()), errors.ECODE_INVAL)
408

    
409
  dirs = [(constants.RUN_GANETI_DIR, constants.RUN_DIRS_MODE)]
410
  utils.EnsureDirs(dirs)
411

    
412
  utils.ForceDictType(beparams, constants.BES_PARAMETER_TYPES)
413
  utils.ForceDictType(nicparams, constants.NICS_PARAMETER_TYPES)
414
  objects.NIC.CheckParameterSyntax(nicparams)
415

    
416
  if ndparams is not None:
417
    utils.ForceDictType(ndparams, constants.NDS_PARAMETER_TYPES)
418
  else:
419
    ndparams = dict(constants.NDC_DEFAULTS)
420

    
421
  # hvparams is a mapping of hypervisor->hvparams dict
422
  for hv_name, hv_params in hvparams.iteritems():
423
    utils.ForceDictType(hv_params, constants.HVS_PARAMETER_TYPES)
424
    hv_class = hypervisor.GetHypervisor(hv_name)
425
    hv_class.CheckParameterSyntax(hv_params)
426

    
427
  # set up ssh config and /etc/hosts
428
  sshline = utils.ReadFile(constants.SSH_HOST_RSA_PUB)
429
  sshkey = sshline.split(" ")[1]
430

    
431
  if modify_etc_hosts:
432
    utils.AddHostToEtcHosts(hostname.name, hostname.ip)
433

    
434
  if modify_ssh_setup:
435
    _InitSSHSetup()
436

    
437
  if default_iallocator is not None:
438
    alloc_script = utils.FindFile(default_iallocator,
439
                                  constants.IALLOCATOR_SEARCH_PATH,
440
                                  os.path.isfile)
441
    if alloc_script is None:
442
      raise errors.OpPrereqError("Invalid default iallocator script '%s'"
443
                                 " specified" % default_iallocator,
444
                                 errors.ECODE_INVAL)
445
  elif constants.HTOOLS:
446
    # htools was enabled at build-time, we default to it
447
    if utils.FindFile(constants.IALLOC_HAIL,
448
                      constants.IALLOCATOR_SEARCH_PATH,
449
                      os.path.isfile):
450
      default_iallocator = constants.IALLOC_HAIL
451

    
452
  now = time.time()
453

    
454
  # init of cluster config file
455
  cluster_config = objects.Cluster(
456
    serial_no=1,
457
    rsahostkeypub=sshkey,
458
    highest_used_port=(constants.FIRST_DRBD_PORT - 1),
459
    mac_prefix=mac_prefix,
460
    volume_group_name=vg_name,
461
    tcpudp_port_pool=set(),
462
    master_node=hostname.name,
463
    master_ip=clustername.ip,
464
    master_netmask=master_netmask,
465
    master_netdev=master_netdev,
466
    cluster_name=clustername.name,
467
    file_storage_dir=file_storage_dir,
468
    shared_file_storage_dir=shared_file_storage_dir,
469
    enabled_hypervisors=enabled_hypervisors,
470
    beparams={constants.PP_DEFAULT: beparams},
471
    nicparams={constants.PP_DEFAULT: nicparams},
472
    ndparams=ndparams,
473
    hvparams=hvparams,
474
    candidate_pool_size=candidate_pool_size,
475
    modify_etc_hosts=modify_etc_hosts,
476
    modify_ssh_setup=modify_ssh_setup,
477
    uid_pool=uid_pool,
478
    ctime=now,
479
    mtime=now,
480
    maintain_node_health=maintain_node_health,
481
    drbd_usermode_helper=drbd_helper,
482
    default_iallocator=default_iallocator,
483
    primary_ip_family=ipcls.family,
484
    prealloc_wipe_disks=prealloc_wipe_disks,
485
    )
486
  master_node_config = objects.Node(name=hostname.name,
487
                                    primary_ip=hostname.ip,
488
                                    secondary_ip=secondary_ip,
489
                                    serial_no=1,
490
                                    master_candidate=True,
491
                                    offline=False, drained=False,
492
                                    ctime=now, mtime=now,
493
                                    )
494
  InitConfig(constants.CONFIG_VERSION, cluster_config, master_node_config)
495
  cfg = config.ConfigWriter(offline=True)
496
  ssh.WriteKnownHostsFile(cfg, constants.SSH_KNOWN_HOSTS_FILE)
497
  cfg.Update(cfg.GetClusterInfo(), logging.error)
498
  backend.WriteSsconfFiles(cfg.GetSsconfValues())
499

    
500
  # set up the inter-node password and certificate
501
  _InitGanetiServerSetup(hostname.name)
502

    
503
  logging.debug("Starting daemons")
504
  result = utils.RunCmd([constants.DAEMON_UTIL, "start-all"])
505
  if result.failed:
506
    raise errors.OpExecError("Could not start daemons, command %s"
507
                             " had exitcode %s and error %s" %
508
                             (result.cmd, result.exit_code, result.output))
509

    
510
  _WaitForMasterDaemon()
511

    
512

    
513
def InitConfig(version, cluster_config, master_node_config,
514
               cfg_file=constants.CLUSTER_CONF_FILE):
515
  """Create the initial cluster configuration.
516

517
  It will contain the current node, which will also be the master
518
  node, and no instances.
519

520
  @type version: int
521
  @param version: configuration version
522
  @type cluster_config: L{objects.Cluster}
523
  @param cluster_config: cluster configuration
524
  @type master_node_config: L{objects.Node}
525
  @param master_node_config: master node configuration
526
  @type cfg_file: string
527
  @param cfg_file: configuration file path
528

529
  """
530
  uuid_generator = config.TemporaryReservationManager()
531
  cluster_config.uuid = uuid_generator.Generate([], utils.NewUUID,
532
                                                _INITCONF_ECID)
533
  master_node_config.uuid = uuid_generator.Generate([], utils.NewUUID,
534
                                                    _INITCONF_ECID)
535
  nodes = {
536
    master_node_config.name: master_node_config,
537
    }
538
  default_nodegroup = objects.NodeGroup(
539
    uuid=uuid_generator.Generate([], utils.NewUUID, _INITCONF_ECID),
540
    name=constants.INITIAL_NODE_GROUP_NAME,
541
    members=[master_node_config.name],
542
    )
543
  nodegroups = {
544
    default_nodegroup.uuid: default_nodegroup,
545
    }
546
  now = time.time()
547
  config_data = objects.ConfigData(version=version,
548
                                   cluster=cluster_config,
549
                                   nodegroups=nodegroups,
550
                                   nodes=nodes,
551
                                   instances={},
552
                                   serial_no=1,
553
                                   ctime=now, mtime=now)
554
  utils.WriteFile(cfg_file,
555
                  data=serializer.Dump(config_data.ToDict()),
556
                  mode=0600)
557

    
558

    
559
def FinalizeClusterDestroy(master):
560
  """Execute the last steps of cluster destroy
561

562
  This function shuts down all the daemons, completing the destroy
563
  begun in cmdlib.LUDestroyOpcode.
564

565
  """
566
  cfg = config.ConfigWriter()
567
  modify_ssh_setup = cfg.GetClusterInfo().modify_ssh_setup
568
  runner = rpc.BootstrapRunner()
569

    
570
  result = runner.call_node_deactivate_master_ip(master)
571
  msg = result.fail_msg
572
  if msg:
573
    logging.warning("Could not disable the master IP: %s", msg)
574

    
575
  result = runner.call_node_stop_master(master)
576
  msg = result.fail_msg
577
  if msg:
578
    logging.warning("Could not disable the master role: %s", msg)
579

    
580
  result = runner.call_node_leave_cluster(master, modify_ssh_setup)
581
  msg = result.fail_msg
582
  if msg:
583
    logging.warning("Could not shutdown the node daemon and cleanup"
584
                    " the node: %s", msg)
585

    
586

    
587
def SetupNodeDaemon(cluster_name, node, ssh_key_check):
588
  """Add a node to the cluster.
589

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

594
  @param cluster_name: the cluster name
595
  @param node: the name of the new node
596
  @param ssh_key_check: whether to do a strict key check
597

598
  """
599
  family = ssconf.SimpleStore().GetPrimaryIPFamily()
600
  sshrunner = ssh.SshRunner(cluster_name,
601
                            ipv6=(family == netutils.IP6Address.family))
602

    
603
  bind_address = constants.IP4_ADDRESS_ANY
604
  if family == netutils.IP6Address.family:
605
    bind_address = constants.IP6_ADDRESS_ANY
606

    
607
  # set up inter-node password and certificate and restarts the node daemon
608
  # and then connect with ssh to set password and start ganeti-noded
609
  # note that all the below variables are sanitized at this point,
610
  # either by being constants or by the checks above
611
  sshrunner.CopyFileToNode(node, constants.NODED_CERT_FILE)
612
  sshrunner.CopyFileToNode(node, constants.RAPI_CERT_FILE)
613
  sshrunner.CopyFileToNode(node, constants.SPICE_CERT_FILE)
614
  sshrunner.CopyFileToNode(node, constants.SPICE_CACERT_FILE)
615
  sshrunner.CopyFileToNode(node, constants.CONFD_HMAC_KEY)
616
  mycommand = ("%s stop-all; %s start %s -b %s" %
617
               (constants.DAEMON_UTIL, constants.DAEMON_UTIL, constants.NODED,
618
                utils.ShellQuote(bind_address)))
619

    
620
  result = sshrunner.Run(node, 'root', mycommand, batch=False,
621
                         ask_key=ssh_key_check,
622
                         use_cluster_key=True,
623
                         strict_host_check=ssh_key_check)
624
  if result.failed:
625
    raise errors.OpExecError("Remote command on node %s, error: %s,"
626
                             " output: %s" %
627
                             (node, result.fail_reason, result.output))
628

    
629
  _WaitForNodeDaemon(node)
630

    
631

    
632
def MasterFailover(no_voting=False):
633
  """Failover the master node.
634

635
  This checks that we are not already the master, and will cause the
636
  current master to cease being master, and the non-master to become
637
  new master.
638

639
  @type no_voting: boolean
640
  @param no_voting: force the operation without remote nodes agreement
641
                      (dangerous)
642

643
  """
644
  sstore = ssconf.SimpleStore()
645

    
646
  old_master, new_master = ssconf.GetMasterAndMyself(sstore)
647
  node_list = sstore.GetNodeList()
648
  mc_list = sstore.GetMasterCandidates()
649

    
650
  if old_master == new_master:
651
    raise errors.OpPrereqError("This commands must be run on the node"
652
                               " where you want the new master to be."
653
                               " %s is already the master" %
654
                               old_master, errors.ECODE_INVAL)
655

    
656
  if new_master not in mc_list:
657
    mc_no_master = [name for name in mc_list if name != old_master]
658
    raise errors.OpPrereqError("This node is not among the nodes marked"
659
                               " as master candidates. Only these nodes"
660
                               " can become masters. Current list of"
661
                               " master candidates is:\n"
662
                               "%s" % ('\n'.join(mc_no_master)),
663
                               errors.ECODE_STATE)
664

    
665
  if not no_voting:
666
    vote_list = GatherMasterVotes(node_list)
667

    
668
    if vote_list:
669
      voted_master = vote_list[0][0]
670
      if voted_master is None:
671
        raise errors.OpPrereqError("Cluster is inconsistent, most nodes did"
672
                                   " not respond.", errors.ECODE_ENVIRON)
673
      elif voted_master != old_master:
674
        raise errors.OpPrereqError("I have a wrong configuration, I believe"
675
                                   " the master is %s but the other nodes"
676
                                   " voted %s. Please resync the configuration"
677
                                   " of this node." %
678
                                   (old_master, voted_master),
679
                                   errors.ECODE_STATE)
680
  # end checks
681

    
682
  rcode = 0
683

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

    
686
  try:
687
    # instantiate a real config writer, as we now know we have the
688
    # configuration data
689
    cfg = config.ConfigWriter(accept_foreign=True)
690

    
691
    cluster_info = cfg.GetClusterInfo()
692
    cluster_info.master_node = new_master
693
    # this will also regenerate the ssconf files, since we updated the
694
    # cluster info
695
    cfg.Update(cluster_info, logging.error)
696
  except errors.ConfigurationError, err:
697
    logging.error("Error while trying to set the new master: %s",
698
                  str(err))
699
    return 1
700

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

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

    
708
  runner = rpc.BootstrapRunner()
709
  result = runner.call_node_deactivate_master_ip(old_master)
710
  msg = result.fail_msg
711
  if msg:
712
    logging.warning("Could not disable the master IP: %s", msg)
713

    
714
  result = runner.call_node_stop_master(old_master)
715
  msg = result.fail_msg
716
  if msg:
717
    logging.error("Could not disable the master role on the old master"
718
                 " %s, please disable manually: %s", old_master, msg)
719

    
720
  logging.info("Checking master IP non-reachability...")
721

    
722
  master_ip = sstore.GetMasterIP()
723
  total_timeout = 30
724

    
725
  # Here we have a phase where no master should be running
726
  def _check_ip():
727
    if netutils.TcpPing(master_ip, constants.DEFAULT_NODED_PORT):
728
      raise utils.RetryAgain()
729

    
730
  try:
731
    utils.Retry(_check_ip, (1, 1.5, 5), total_timeout)
732
  except utils.RetryTimeout:
733
    logging.warning("The master IP is still reachable after %s seconds,"
734
                    " continuing but activating the master on the current"
735
                    " node will probably fail", total_timeout)
736

    
737
  if jstore.CheckDrainFlag():
738
    logging.info("Undraining job queue")
739
    jstore.SetDrainFlag(False)
740

    
741
  logging.info("Starting the master daemons on the new master")
742

    
743
  result = rpc.BootstrapRunner().call_node_start_master_daemons(new_master,
744
                                                                no_voting)
745
  msg = result.fail_msg
746
  if msg:
747
    logging.error("Could not start the master role on the new master"
748
                  " %s, please check: %s", new_master, msg)
749
    rcode = 1
750

    
751
  logging.info("Master failed over from %s to %s", old_master, new_master)
752
  return rcode
753

    
754

    
755
def GetMaster():
756
  """Returns the current master node.
757

758
  This is a separate function in bootstrap since it's needed by
759
  gnt-cluster, and instead of importing directly ssconf, it's better
760
  to abstract it in bootstrap, where we do use ssconf in other
761
  functions too.
762

763
  """
764
  sstore = ssconf.SimpleStore()
765

    
766
  old_master, _ = ssconf.GetMasterAndMyself(sstore)
767

    
768
  return old_master
769

    
770

    
771
def GatherMasterVotes(node_list):
772
  """Check the agreement on who is the master.
773

774
  This function will return a list of (node, number of votes), ordered
775
  by the number of votes. Errors will be denoted by the key 'None'.
776

777
  Note that the sum of votes is the number of nodes this machine
778
  knows, whereas the number of entries in the list could be different
779
  (if some nodes vote for another master).
780

781
  We remove ourselves from the list since we know that (bugs aside)
782
  since we use the same source for configuration information for both
783
  backend and boostrap, we'll always vote for ourselves.
784

785
  @type node_list: list
786
  @param node_list: the list of nodes to query for master info; the current
787
      node will be removed if it is in the list
788
  @rtype: list
789
  @return: list of (node, votes)
790

791
  """
792
  myself = netutils.Hostname.GetSysName()
793
  try:
794
    node_list.remove(myself)
795
  except ValueError:
796
    pass
797
  if not node_list:
798
    # no nodes left (eventually after removing myself)
799
    return []
800
  results = rpc.BootstrapRunner().call_master_info(node_list)
801
  if not isinstance(results, dict):
802
    # this should not happen (unless internal error in rpc)
803
    logging.critical("Can't complete rpc call, aborting master startup")
804
    return [(None, len(node_list))]
805
  votes = {}
806
  for node in results:
807
    nres = results[node]
808
    data = nres.payload
809
    msg = nres.fail_msg
810
    fail = False
811
    if msg:
812
      logging.warning("Error contacting node %s: %s", node, msg)
813
      fail = True
814
    # for now we accept both length 3, 4 and 5 (data[3] is primary ip version
815
    # and data[4] is the master netmask)
816
    elif not isinstance(data, (tuple, list)) or len(data) < 3:
817
      logging.warning("Invalid data received from node %s: %s", node, data)
818
      fail = True
819
    if fail:
820
      if None not in votes:
821
        votes[None] = 0
822
      votes[None] += 1
823
      continue
824
    master_node = data[2]
825
    if master_node not in votes:
826
      votes[master_node] = 0
827
    votes[master_node] += 1
828

    
829
  vote_list = [v for v in votes.items()]
830
  # sort first on number of votes then on name, since we want None
831
  # sorted later if we have the half of the nodes not responding, and
832
  # half voting all for the same master
833
  vote_list.sort(key=lambda x: (x[1], x[0]), reverse=True)
834

    
835
  return vote_list