Statistics
| Branch: | Tag: | Revision:

root / lib / bootstrap.py @ 75cf411a

History | View | Annotate | Download (27.3 kB)

1
#
2
#
3

    
4
# Copyright (C) 2006, 2007, 2008, 2010 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

    
46
# ec_id for InitConfig's temporary reservation manager
47
_INITCONF_ECID = "initconfig-ecid"
48

    
49

    
50
def _InitSSHSetup():
51
  """Setup the SSH configuration for the cluster.
52

53
  This generates a dsa keypair for root, adds the pub key to the
54
  permitted hosts and adds the hostkey to its own known hosts.
55

56
  """
57
  priv_key, pub_key, auth_keys = ssh.GetUserFiles(constants.GANETI_RUNAS)
58

    
59
  for name in priv_key, pub_key:
60
    if os.path.exists(name):
61
      utils.CreateBackup(name)
62
    utils.RemoveFile(name)
63

    
64
  result = utils.RunCmd(["ssh-keygen", "-t", "dsa",
65
                         "-f", priv_key,
66
                         "-q", "-N", ""])
67
  if result.failed:
68
    raise errors.OpExecError("Could not generate ssh keypair, error %s" %
69
                             result.output)
70

    
71
  utils.AddAuthorizedKey(auth_keys, utils.ReadFile(pub_key))
72

    
73

    
74
def GenerateHmacKey(file_name):
75
  """Writes a new HMAC key.
76

77
  @type file_name: str
78
  @param file_name: Path to output file
79

80
  """
81
  utils.WriteFile(file_name, data="%s\n" % utils.GenerateSecret(), mode=0400,
82
                  backup=True)
83

    
84

    
85
def GenerateClusterCrypto(new_cluster_cert, new_rapi_cert, new_confd_hmac_key,
86
                          new_cds, rapi_cert_pem=None, cds=None,
87
                          nodecert_file=constants.NODED_CERT_FILE,
88
                          rapicert_file=constants.RAPI_CERT_FILE,
89
                          hmackey_file=constants.CONFD_HMAC_KEY,
90
                          cds_file=constants.CLUSTER_DOMAIN_SECRET_FILE):
91
  """Updates the cluster certificates, keys and secrets.
92

93
  @type new_cluster_cert: bool
94
  @param new_cluster_cert: Whether to generate a new cluster certificate
95
  @type new_rapi_cert: bool
96
  @param new_rapi_cert: Whether to generate a new RAPI certificate
97
  @type new_confd_hmac_key: bool
98
  @param new_confd_hmac_key: Whether to generate a new HMAC key
99
  @type new_cds: bool
100
  @param new_cds: Whether to generate a new cluster domain secret
101
  @type rapi_cert_pem: string
102
  @param rapi_cert_pem: New RAPI certificate in PEM format
103
  @type cds: string
104
  @param cds: New cluster domain secret
105
  @type nodecert_file: string
106
  @param nodecert_file: optional override of the node cert file path
107
  @type rapicert_file: string
108
  @param rapicert_file: optional override of the rapi cert file path
109
  @type hmackey_file: string
110
  @param hmackey_file: optional override of the hmac key file path
111

112
  """
113
  # noded SSL certificate
114
  cluster_cert_exists = os.path.exists(nodecert_file)
115
  if new_cluster_cert or not cluster_cert_exists:
116
    if cluster_cert_exists:
117
      utils.CreateBackup(nodecert_file)
118

    
119
    logging.debug("Generating new cluster certificate at %s", nodecert_file)
120
    utils.GenerateSelfSignedSslCert(nodecert_file)
121

    
122
  # confd HMAC key
123
  if new_confd_hmac_key or not os.path.exists(hmackey_file):
124
    logging.debug("Writing new confd HMAC key to %s", hmackey_file)
125
    GenerateHmacKey(hmackey_file)
126

    
127
  # RAPI
128
  rapi_cert_exists = os.path.exists(rapicert_file)
129

    
130
  if rapi_cert_pem:
131
    # Assume rapi_pem contains a valid PEM-formatted certificate and key
132
    logging.debug("Writing RAPI certificate at %s", rapicert_file)
133
    utils.WriteFile(rapicert_file, data=rapi_cert_pem, backup=True)
134

    
135
  elif new_rapi_cert or not rapi_cert_exists:
136
    if rapi_cert_exists:
137
      utils.CreateBackup(rapicert_file)
138

    
139
    logging.debug("Generating new RAPI certificate at %s", rapicert_file)
140
    utils.GenerateSelfSignedSslCert(rapicert_file)
141

    
142
  # Cluster domain secret
143
  if cds:
144
    logging.debug("Writing cluster domain secret to %s", cds_file)
145
    utils.WriteFile(cds_file, data=cds, backup=True)
146

    
147
  elif new_cds or not os.path.exists(cds_file):
148
    logging.debug("Generating new cluster domain secret at %s", cds_file)
149
    GenerateHmacKey(cds_file)
150

    
151

    
152
def _InitGanetiServerSetup(master_name):
153
  """Setup the necessary configuration for the initial node daemon.
154

155
  This creates the nodepass file containing the shared password for
156
  the cluster, generates the SSL certificate and starts the node daemon.
157

158
  @type master_name: str
159
  @param master_name: Name of the master node
160

161
  """
162
  # Generate cluster secrets
163
  GenerateClusterCrypto(True, False, False, False)
164

    
165
  result = utils.RunCmd([constants.DAEMON_UTIL, "start", constants.NODED])
166
  if result.failed:
167
    raise errors.OpExecError("Could not start the node daemon, command %s"
168
                             " had exitcode %s and error %s" %
169
                             (result.cmd, result.exit_code, result.output))
170

    
171
  _WaitForNodeDaemon(master_name)
172

    
173

    
174
def _WaitForNodeDaemon(node_name):
175
  """Wait for node daemon to become responsive.
176

177
  """
178
  def _CheckNodeDaemon():
179
    result = rpc.RpcRunner.call_version([node_name])[node_name]
180
    if result.fail_msg:
181
      raise utils.RetryAgain()
182

    
183
  try:
184
    utils.Retry(_CheckNodeDaemon, 1.0, 10.0)
185
  except utils.RetryTimeout:
186
    raise errors.OpExecError("Node daemon on %s didn't answer queries within"
187
                             " 10 seconds" % node_name)
188

    
189

    
190
def _InitFileStorage(file_storage_dir):
191
  """Initialize if needed the file storage.
192

193
  @param file_storage_dir: the user-supplied value
194
  @return: either empty string (if file storage was disabled at build
195
      time) or the normalized path to the storage directory
196

197
  """
198
  if not constants.ENABLE_FILE_STORAGE:
199
    return ""
200

    
201
  file_storage_dir = os.path.normpath(file_storage_dir)
202

    
203
  if not os.path.isabs(file_storage_dir):
204
    raise errors.OpPrereqError("The file storage directory you passed is"
205
                               " not an absolute path.", errors.ECODE_INVAL)
206

    
207
  if not os.path.exists(file_storage_dir):
208
    try:
209
      os.makedirs(file_storage_dir, 0750)
210
    except OSError, err:
211
      raise errors.OpPrereqError("Cannot create file storage directory"
212
                                 " '%s': %s" % (file_storage_dir, err),
213
                                 errors.ECODE_ENVIRON)
214

    
215
  if not os.path.isdir(file_storage_dir):
216
    raise errors.OpPrereqError("The file storage directory '%s' is not"
217
                               " a directory." % file_storage_dir,
218
                               errors.ECODE_ENVIRON)
219
  return file_storage_dir
220

    
221

    
222
#pylint: disable-msg=R0913
223
def InitCluster(cluster_name, mac_prefix,
224
                master_netdev, file_storage_dir, candidate_pool_size,
225
                secondary_ip=None, vg_name=None, beparams=None,
226
                nicparams=None, ndparams=None, hvparams=None,
227
                enabled_hypervisors=None, modify_etc_hosts=True,
228
                modify_ssh_setup=True, maintain_node_health=False,
229
                drbd_helper=None, uid_pool=None, default_iallocator=None,
230
                primary_ip_version=None, prealloc_wipe_disks=False):
231
  """Initialise the cluster.
232

233
  @type candidate_pool_size: int
234
  @param candidate_pool_size: master candidate pool size
235

236
  """
237
  # TODO: complete the docstring
238
  if config.ConfigWriter.IsCluster():
239
    raise errors.OpPrereqError("Cluster is already initialised",
240
                               errors.ECODE_STATE)
241

    
242
  if not enabled_hypervisors:
243
    raise errors.OpPrereqError("Enabled hypervisors list must contain at"
244
                               " least one member", errors.ECODE_INVAL)
245
  invalid_hvs = set(enabled_hypervisors) - constants.HYPER_TYPES
246
  if invalid_hvs:
247
    raise errors.OpPrereqError("Enabled hypervisors contains invalid"
248
                               " entries: %s" % invalid_hvs,
249
                               errors.ECODE_INVAL)
250

    
251

    
252
  ipcls = None
253
  if primary_ip_version == constants.IP4_VERSION:
254
    ipcls = netutils.IP4Address
255
  elif primary_ip_version == constants.IP6_VERSION:
256
    ipcls = netutils.IP6Address
257
  else:
258
    raise errors.OpPrereqError("Invalid primary ip version: %d." %
259
                               primary_ip_version)
260

    
261
  hostname = netutils.GetHostname(family=ipcls.family)
262
  if not ipcls.IsValid(hostname.ip):
263
    raise errors.OpPrereqError("This host's IP (%s) is not a valid IPv%d"
264
                               " address." % (hostname.ip, primary_ip_version))
265

    
266
  if ipcls.IsLoopback(hostname.ip):
267
    raise errors.OpPrereqError("This host's IP (%s) resolves to a loopback"
268
                               " address. Please fix DNS or %s." %
269
                               (hostname.ip, constants.ETC_HOSTS),
270
                               errors.ECODE_ENVIRON)
271

    
272
  if not ipcls.Own(hostname.ip):
273
    raise errors.OpPrereqError("Inconsistency: this host's name resolves"
274
                               " to %s,\nbut this ip address does not"
275
                               " belong to this host" %
276
                               hostname.ip, errors.ECODE_ENVIRON)
277

    
278
  clustername = netutils.GetHostname(name=cluster_name, family=ipcls.family)
279

    
280
  if netutils.TcpPing(clustername.ip, constants.DEFAULT_NODED_PORT, timeout=5):
281
    raise errors.OpPrereqError("Cluster IP already active",
282
                               errors.ECODE_NOTUNIQUE)
283

    
284
  if not secondary_ip:
285
    if primary_ip_version == constants.IP6_VERSION:
286
      raise errors.OpPrereqError("When using a IPv6 primary address, a valid"
287
                                 " IPv4 address must be given as secondary",
288
                                 errors.ECODE_INVAL)
289
    secondary_ip = hostname.ip
290

    
291
  if not netutils.IP4Address.IsValid(secondary_ip):
292
    raise errors.OpPrereqError("Secondary IP address (%s) has to be a valid"
293
                               " IPv4 address." % secondary_ip,
294
                               errors.ECODE_INVAL)
295

    
296
  if not netutils.IP4Address.Own(secondary_ip):
297
    raise errors.OpPrereqError("You gave %s as secondary IP,"
298
                               " but it does not belong to this host." %
299
                               secondary_ip, errors.ECODE_ENVIRON)
300

    
301
  if vg_name is not None:
302
    # Check if volume group is valid
303
    vgstatus = utils.CheckVolumeGroupSize(utils.ListVolumeGroups(), vg_name,
304
                                          constants.MIN_VG_SIZE)
305
    if vgstatus:
306
      raise errors.OpPrereqError("Error: %s\nspecify --no-lvm-storage if"
307
                                 " you are not using lvm" % vgstatus,
308
                                 errors.ECODE_INVAL)
309

    
310
  if drbd_helper is not None:
311
    try:
312
      curr_helper = bdev.BaseDRBD.GetUsermodeHelper()
313
    except errors.BlockDeviceError, err:
314
      raise errors.OpPrereqError("Error while checking drbd helper"
315
                                 " (specify --no-drbd-storage if you are not"
316
                                 " using drbd): %s" % str(err),
317
                                 errors.ECODE_ENVIRON)
318
    if drbd_helper != curr_helper:
319
      raise errors.OpPrereqError("Error: requiring %s as drbd helper but %s"
320
                                 " is the current helper" % (drbd_helper,
321
                                                             curr_helper),
322
                                 errors.ECODE_INVAL)
323

    
324
  file_storage_dir = _InitFileStorage(file_storage_dir)
325

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

    
330
  result = utils.RunCmd(["ip", "link", "show", "dev", master_netdev])
331
  if result.failed:
332
    raise errors.OpPrereqError("Invalid master netdev given (%s): '%s'" %
333
                               (master_netdev,
334
                                result.output.strip()), errors.ECODE_INVAL)
335

    
336
  dirs = [(constants.RUN_GANETI_DIR, constants.RUN_DIRS_MODE)]
337
  utils.EnsureDirs(dirs)
338

    
339
  utils.ForceDictType(beparams, constants.BES_PARAMETER_TYPES)
340
  utils.ForceDictType(nicparams, constants.NICS_PARAMETER_TYPES)
341
  objects.NIC.CheckParameterSyntax(nicparams)
342

    
343
  if ndparams is not None:
344
    utils.ForceDictType(ndparams, constants.NDS_PARAMETER_TYPES)
345
  else:
346
    ndparams = dict(constants.NDC_DEFAULTS)
347

    
348
  # hvparams is a mapping of hypervisor->hvparams dict
349
  for hv_name, hv_params in hvparams.iteritems():
350
    utils.ForceDictType(hv_params, constants.HVS_PARAMETER_TYPES)
351
    hv_class = hypervisor.GetHypervisor(hv_name)
352
    hv_class.CheckParameterSyntax(hv_params)
353

    
354
  # set up ssh config and /etc/hosts
355
  sshline = utils.ReadFile(constants.SSH_HOST_RSA_PUB)
356
  sshkey = sshline.split(" ")[1]
357

    
358
  if modify_etc_hosts:
359
    utils.AddHostToEtcHosts(hostname.name, hostname.ip)
360

    
361
  if modify_ssh_setup:
362
    _InitSSHSetup()
363

    
364
  if default_iallocator is not None:
365
    alloc_script = utils.FindFile(default_iallocator,
366
                                  constants.IALLOCATOR_SEARCH_PATH,
367
                                  os.path.isfile)
368
    if alloc_script is None:
369
      raise errors.OpPrereqError("Invalid default iallocator script '%s'"
370
                                 " specified" % default_iallocator,
371
                                 errors.ECODE_INVAL)
372

    
373
  now = time.time()
374

    
375
  # init of cluster config file
376
  cluster_config = objects.Cluster(
377
    serial_no=1,
378
    rsahostkeypub=sshkey,
379
    highest_used_port=(constants.FIRST_DRBD_PORT - 1),
380
    mac_prefix=mac_prefix,
381
    volume_group_name=vg_name,
382
    tcpudp_port_pool=set(),
383
    master_node=hostname.name,
384
    master_ip=clustername.ip,
385
    master_netdev=master_netdev,
386
    cluster_name=clustername.name,
387
    file_storage_dir=file_storage_dir,
388
    enabled_hypervisors=enabled_hypervisors,
389
    beparams={constants.PP_DEFAULT: beparams},
390
    nicparams={constants.PP_DEFAULT: nicparams},
391
    ndparams=ndparams,
392
    hvparams=hvparams,
393
    candidate_pool_size=candidate_pool_size,
394
    modify_etc_hosts=modify_etc_hosts,
395
    modify_ssh_setup=modify_ssh_setup,
396
    uid_pool=uid_pool,
397
    ctime=now,
398
    mtime=now,
399
    maintain_node_health=maintain_node_health,
400
    drbd_usermode_helper=drbd_helper,
401
    default_iallocator=default_iallocator,
402
    primary_ip_family=ipcls.family,
403
    prealloc_wipe_disks=prealloc_wipe_disks,
404
    )
405
  master_node_config = objects.Node(name=hostname.name,
406
                                    primary_ip=hostname.ip,
407
                                    secondary_ip=secondary_ip,
408
                                    serial_no=1,
409
                                    master_candidate=True,
410
                                    offline=False, drained=False,
411
                                    ctime=now, mtime=now,
412
                                    )
413
  InitConfig(constants.CONFIG_VERSION, cluster_config, master_node_config)
414
  cfg = config.ConfigWriter(offline=True)
415
  ssh.WriteKnownHostsFile(cfg, constants.SSH_KNOWN_HOSTS_FILE)
416
  cfg.Update(cfg.GetClusterInfo(), logging.error)
417
  backend.WriteSsconfFiles(cfg.GetSsconfValues())
418

    
419
  # set up the inter-node password and certificate
420
  _InitGanetiServerSetup(hostname.name)
421

    
422
  # start the master ip
423
  # TODO: Review rpc call from bootstrap
424
  # TODO: Warn on failed start master
425
  rpc.RpcRunner.call_node_start_master(hostname.name, True, False)
426

    
427

    
428
def InitConfig(version, cluster_config, master_node_config,
429
               cfg_file=constants.CLUSTER_CONF_FILE):
430
  """Create the initial cluster configuration.
431

432
  It will contain the current node, which will also be the master
433
  node, and no instances.
434

435
  @type version: int
436
  @param version: configuration version
437
  @type cluster_config: L{objects.Cluster}
438
  @param cluster_config: cluster configuration
439
  @type master_node_config: L{objects.Node}
440
  @param master_node_config: master node configuration
441
  @type cfg_file: string
442
  @param cfg_file: configuration file path
443

444
  """
445
  uuid_generator = config.TemporaryReservationManager()
446
  cluster_config.uuid = uuid_generator.Generate([], utils.NewUUID,
447
                                                _INITCONF_ECID)
448
  master_node_config.uuid = uuid_generator.Generate([], utils.NewUUID,
449
                                                    _INITCONF_ECID)
450
  nodes = {
451
    master_node_config.name: master_node_config,
452
    }
453
  default_nodegroup = objects.NodeGroup(
454
    uuid=uuid_generator.Generate([], utils.NewUUID, _INITCONF_ECID),
455
    name=constants.INITIAL_NODE_GROUP_NAME,
456
    members=[master_node_config.name],
457
    )
458
  nodegroups = {
459
    default_nodegroup.uuid: default_nodegroup,
460
    }
461
  now = time.time()
462
  config_data = objects.ConfigData(version=version,
463
                                   cluster=cluster_config,
464
                                   nodegroups=nodegroups,
465
                                   nodes=nodes,
466
                                   instances={},
467
                                   serial_no=1,
468
                                   ctime=now, mtime=now)
469
  utils.WriteFile(cfg_file,
470
                  data=serializer.Dump(config_data.ToDict()),
471
                  mode=0600)
472

    
473

    
474
def FinalizeClusterDestroy(master):
475
  """Execute the last steps of cluster destroy
476

477
  This function shuts down all the daemons, completing the destroy
478
  begun in cmdlib.LUDestroyOpcode.
479

480
  """
481
  cfg = config.ConfigWriter()
482
  modify_ssh_setup = cfg.GetClusterInfo().modify_ssh_setup
483
  result = rpc.RpcRunner.call_node_stop_master(master, True)
484
  msg = result.fail_msg
485
  if msg:
486
    logging.warning("Could not disable the master role: %s", msg)
487
  result = rpc.RpcRunner.call_node_leave_cluster(master, modify_ssh_setup)
488
  msg = result.fail_msg
489
  if msg:
490
    logging.warning("Could not shutdown the node daemon and cleanup"
491
                    " the node: %s", msg)
492

    
493

    
494
def SetupNodeDaemon(cluster_name, node, ssh_key_check):
495
  """Add a node to the cluster.
496

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

501
  @param cluster_name: the cluster name
502
  @param node: the name of the new node
503
  @param ssh_key_check: whether to do a strict key check
504

505
  """
506
  family = ssconf.SimpleStore().GetPrimaryIPFamily()
507
  sshrunner = ssh.SshRunner(cluster_name,
508
                            ipv6=family==netutils.IP6Address.family)
509

    
510
  noded_cert = utils.ReadFile(constants.NODED_CERT_FILE)
511
  rapi_cert = utils.ReadFile(constants.RAPI_CERT_FILE)
512
  confd_hmac_key = utils.ReadFile(constants.CONFD_HMAC_KEY)
513

    
514
  # in the base64 pem encoding, neither '!' nor '.' are valid chars,
515
  # so we use this to detect an invalid certificate; as long as the
516
  # cert doesn't contain this, the here-document will be correctly
517
  # parsed by the shell sequence below. HMAC keys are hexadecimal strings,
518
  # so the same restrictions apply.
519
  for content in (noded_cert, rapi_cert, confd_hmac_key):
520
    if re.search('^!EOF\.', content, re.MULTILINE):
521
      raise errors.OpExecError("invalid SSL certificate or HMAC key")
522

    
523
  if not noded_cert.endswith("\n"):
524
    noded_cert += "\n"
525
  if not rapi_cert.endswith("\n"):
526
    rapi_cert += "\n"
527
  if not confd_hmac_key.endswith("\n"):
528
    confd_hmac_key += "\n"
529

    
530
  bind_address = constants.IP4_ADDRESS_ANY
531
  if family == netutils.IP6Address.family:
532
    bind_address = constants.IP6_ADDRESS_ANY
533

    
534
  # set up inter-node password and certificate and restarts the node daemon
535
  # and then connect with ssh to set password and start ganeti-noded
536
  # note that all the below variables are sanitized at this point,
537
  # either by being constants or by the checks above
538
  sshrunner.CopyFileToNode(node, constants.NODED_CERT_FILE)
539
  sshrunner.CopyFileToNode(node, constants.RAPI_CERT_FILE)
540
  sshrunner.CopyFileToNode(node, constants.CONFD_HMAC_KEY)
541
  mycommand = ("%s stop-all; %s start %s -b '%s'" % (constants.DAEMON_UTIL,
542
                                                     constants.DAEMON_UTIL,
543
                                                     constants.NODED,
544
                                                     bind_address))
545

    
546
  result = sshrunner.Run(node, 'root', mycommand, batch=False,
547
                         ask_key=ssh_key_check,
548
                         use_cluster_key=True,
549
                         strict_host_check=ssh_key_check)
550
  if result.failed:
551
    raise errors.OpExecError("Remote command on node %s, error: %s,"
552
                             " output: %s" %
553
                             (node, result.fail_reason, result.output))
554

    
555
  _WaitForNodeDaemon(node)
556

    
557

    
558
def MasterFailover(no_voting=False):
559
  """Failover the master node.
560

561
  This checks that we are not already the master, and will cause the
562
  current master to cease being master, and the non-master to become
563
  new master.
564

565
  @type no_voting: boolean
566
  @param no_voting: force the operation without remote nodes agreement
567
                      (dangerous)
568

569
  """
570
  sstore = ssconf.SimpleStore()
571

    
572
  old_master, new_master = ssconf.GetMasterAndMyself(sstore)
573
  node_list = sstore.GetNodeList()
574
  mc_list = sstore.GetMasterCandidates()
575

    
576
  if old_master == new_master:
577
    raise errors.OpPrereqError("This commands must be run on the node"
578
                               " where you want the new master to be."
579
                               " %s is already the master" %
580
                               old_master, errors.ECODE_INVAL)
581

    
582
  if new_master not in mc_list:
583
    mc_no_master = [name for name in mc_list if name != old_master]
584
    raise errors.OpPrereqError("This node is not among the nodes marked"
585
                               " as master candidates. Only these nodes"
586
                               " can become masters. Current list of"
587
                               " master candidates is:\n"
588
                               "%s" % ('\n'.join(mc_no_master)),
589
                               errors.ECODE_STATE)
590

    
591
  if not no_voting:
592
    vote_list = GatherMasterVotes(node_list)
593

    
594
    if vote_list:
595
      voted_master = vote_list[0][0]
596
      if voted_master is None:
597
        raise errors.OpPrereqError("Cluster is inconsistent, most nodes did"
598
                                   " not respond.", errors.ECODE_ENVIRON)
599
      elif voted_master != old_master:
600
        raise errors.OpPrereqError("I have a wrong configuration, I believe"
601
                                   " the master is %s but the other nodes"
602
                                   " voted %s. Please resync the configuration"
603
                                   " of this node." %
604
                                   (old_master, voted_master),
605
                                   errors.ECODE_STATE)
606
  # end checks
607

    
608
  rcode = 0
609

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

    
612
  try:
613
    # instantiate a real config writer, as we now know we have the
614
    # configuration data
615
    cfg = config.ConfigWriter(accept_foreign=True)
616

    
617
    cluster_info = cfg.GetClusterInfo()
618
    cluster_info.master_node = new_master
619
    # this will also regenerate the ssconf files, since we updated the
620
    # cluster info
621
    cfg.Update(cluster_info, logging.error)
622
  except errors.ConfigurationError, err:
623
    logging.error("Error while trying to set the new master: %s",
624
                  str(err))
625
    return 1
626

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

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

    
634
  result = rpc.RpcRunner.call_node_stop_master(old_master, True)
635
  msg = result.fail_msg
636
  if msg:
637
    logging.error("Could not disable the master role on the old master"
638
                 " %s, please disable manually: %s", old_master, msg)
639

    
640
  logging.info("Checking master IP non-reachability...")
641

    
642
  master_ip = sstore.GetMasterIP()
643
  total_timeout = 30
644
  # Here we have a phase where no master should be running
645
  def _check_ip():
646
    if netutils.TcpPing(master_ip, constants.DEFAULT_NODED_PORT):
647
      raise utils.RetryAgain()
648

    
649
  try:
650
    utils.Retry(_check_ip, (1, 1.5, 5), total_timeout)
651
  except utils.RetryTimeout:
652
    logging.warning("The master IP is still reachable after %s seconds,"
653
                    " continuing but activating the master on the current"
654
                    " node will probably fail", total_timeout)
655

    
656
  logging.info("Starting the master daemons on the new master")
657

    
658
  result = rpc.RpcRunner.call_node_start_master(new_master, True, no_voting)
659
  msg = result.fail_msg
660
  if msg:
661
    logging.error("Could not start the master role on the new master"
662
                  " %s, please check: %s", new_master, msg)
663
    rcode = 1
664

    
665
  logging.info("Master failed over from %s to %s", old_master, new_master)
666
  return rcode
667

    
668

    
669
def GetMaster():
670
  """Returns the current master node.
671

672
  This is a separate function in bootstrap since it's needed by
673
  gnt-cluster, and instead of importing directly ssconf, it's better
674
  to abstract it in bootstrap, where we do use ssconf in other
675
  functions too.
676

677
  """
678
  sstore = ssconf.SimpleStore()
679

    
680
  old_master, _ = ssconf.GetMasterAndMyself(sstore)
681

    
682
  return old_master
683

    
684

    
685
def GatherMasterVotes(node_list):
686
  """Check the agreement on who is the master.
687

688
  This function will return a list of (node, number of votes), ordered
689
  by the number of votes. Errors will be denoted by the key 'None'.
690

691
  Note that the sum of votes is the number of nodes this machine
692
  knows, whereas the number of entries in the list could be different
693
  (if some nodes vote for another master).
694

695
  We remove ourselves from the list since we know that (bugs aside)
696
  since we use the same source for configuration information for both
697
  backend and boostrap, we'll always vote for ourselves.
698

699
  @type node_list: list
700
  @param node_list: the list of nodes to query for master info; the current
701
      node will be removed if it is in the list
702
  @rtype: list
703
  @return: list of (node, votes)
704

705
  """
706
  myself = netutils.Hostname.GetSysName()
707
  try:
708
    node_list.remove(myself)
709
  except ValueError:
710
    pass
711
  if not node_list:
712
    # no nodes left (eventually after removing myself)
713
    return []
714
  results = rpc.RpcRunner.call_master_info(node_list)
715
  if not isinstance(results, dict):
716
    # this should not happen (unless internal error in rpc)
717
    logging.critical("Can't complete rpc call, aborting master startup")
718
    return [(None, len(node_list))]
719
  votes = {}
720
  for node in results:
721
    nres = results[node]
722
    data = nres.payload
723
    msg = nres.fail_msg
724
    fail = False
725
    if msg:
726
      logging.warning("Error contacting node %s: %s", node, msg)
727
      fail = True
728
    # for now we accept both length 3 and 4 (data[3] is primary ip version)
729
    elif not isinstance(data, (tuple, list)) or len(data) < 3:
730
      logging.warning("Invalid data received from node %s: %s", node, data)
731
      fail = True
732
    if fail:
733
      if None not in votes:
734
        votes[None] = 0
735
      votes[None] += 1
736
      continue
737
    master_node = data[2]
738
    if master_node not in votes:
739
      votes[master_node] = 0
740
    votes[master_node] += 1
741

    
742
  vote_list = [v for v in votes.items()]
743
  # sort first on number of votes then on name, since we want None
744
  # sorted later if we have the half of the nodes not responding, and
745
  # half voting all for the same master
746
  vote_list.sort(key=lambda x: (x[1], x[0]), reverse=True)
747

    
748
  return vote_list