Statistics
| Branch: | Tag: | Revision:

root / lib / bootstrap.py @ 0376655e

History | View | Annotate | Download (27.9 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_confd_hmac_key,
92
                          new_cds, rapi_cert_pem=None, cds=None,
93
                          nodecert_file=constants.NODED_CERT_FILE,
94
                          rapicert_file=constants.RAPI_CERT_FILE,
95
                          hmackey_file=constants.CONFD_HMAC_KEY,
96
                          cds_file=constants.CLUSTER_DOMAIN_SECRET_FILE):
97
  """Updates the cluster certificates, keys and secrets.
98

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

118
  """
119
  # noded SSL certificate
120
  cluster_cert_exists = os.path.exists(nodecert_file)
121
  if new_cluster_cert or not cluster_cert_exists:
122
    if cluster_cert_exists:
123
      utils.CreateBackup(nodecert_file)
124

    
125
    logging.debug("Generating new cluster certificate at %s", nodecert_file)
126
    utils.GenerateSelfSignedSslCert(nodecert_file)
127

    
128
  # confd HMAC key
129
  if new_confd_hmac_key or not os.path.exists(hmackey_file):
130
    logging.debug("Writing new confd HMAC key to %s", hmackey_file)
131
    GenerateHmacKey(hmackey_file)
132

    
133
  # RAPI
134
  rapi_cert_exists = os.path.exists(rapicert_file)
135

    
136
  if rapi_cert_pem:
137
    # Assume rapi_pem contains a valid PEM-formatted certificate and key
138
    logging.debug("Writing RAPI certificate at %s", rapicert_file)
139
    utils.WriteFile(rapicert_file, data=rapi_cert_pem, backup=True)
140

    
141
  elif new_rapi_cert or not rapi_cert_exists:
142
    if rapi_cert_exists:
143
      utils.CreateBackup(rapicert_file)
144

    
145
    logging.debug("Generating new RAPI certificate at %s", rapicert_file)
146
    utils.GenerateSelfSignedSslCert(rapicert_file)
147

    
148
  # Cluster domain secret
149
  if cds:
150
    logging.debug("Writing cluster domain secret to %s", cds_file)
151
    utils.WriteFile(cds_file, data=cds, backup=True)
152

    
153
  elif new_cds or not os.path.exists(cds_file):
154
    logging.debug("Generating new cluster domain secret at %s", cds_file)
155
    GenerateHmacKey(cds_file)
156

    
157

    
158
def _InitGanetiServerSetup(master_name):
159
  """Setup the necessary configuration for the initial node daemon.
160

161
  This creates the nodepass file containing the shared password for
162
  the cluster, generates the SSL certificate and starts the node daemon.
163

164
  @type master_name: str
165
  @param master_name: Name of the master node
166

167
  """
168
  # Generate cluster secrets
169
  GenerateClusterCrypto(True, False, False, False)
170

    
171
  result = utils.RunCmd([constants.DAEMON_UTIL, "start", constants.NODED])
172
  if result.failed:
173
    raise errors.OpExecError("Could not start the node daemon, command %s"
174
                             " had exitcode %s and error %s" %
175
                             (result.cmd, result.exit_code, result.output))
176

    
177
  _WaitForNodeDaemon(master_name)
178

    
179

    
180
def _WaitForNodeDaemon(node_name):
181
  """Wait for node daemon to become responsive.
182

183
  """
184
  def _CheckNodeDaemon():
185
    result = rpc.RpcRunner.call_version([node_name])[node_name]
186
    if result.fail_msg:
187
      raise utils.RetryAgain()
188

    
189
  try:
190
    utils.Retry(_CheckNodeDaemon, 1.0, _DAEMON_READY_TIMEOUT)
191
  except utils.RetryTimeout:
192
    raise errors.OpExecError("Node daemon on %s didn't answer queries within"
193
                             " %s seconds" % (node_name, _DAEMON_READY_TIMEOUT))
194

    
195

    
196
def _WaitForMasterDaemon():
197
  """Wait for master daemon to become responsive.
198

199
  """
200
  def _CheckMasterDaemon():
201
    try:
202
      cl = luxi.Client()
203
      (cluster_name, ) = cl.QueryConfigValues(["cluster_name"])
204
    except Exception:
205
      raise utils.RetryAgain()
206

    
207
    logging.debug("Received cluster name %s from master", cluster_name)
208

    
209
  try:
210
    utils.Retry(_CheckMasterDaemon, 1.0, _DAEMON_READY_TIMEOUT)
211
  except utils.RetryTimeout:
212
    raise errors.OpExecError("Master daemon didn't answer queries within"
213
                             " %s seconds" % _DAEMON_READY_TIMEOUT)
214

    
215

    
216
def _InitFileStorage(file_storage_dir):
217
  """Initialize if needed the file storage.
218

219
  @param file_storage_dir: the user-supplied value
220
  @return: either empty string (if file storage was disabled at build
221
      time) or the normalized path to the storage directory
222

223
  """
224
  file_storage_dir = os.path.normpath(file_storage_dir)
225

    
226
  if not os.path.isabs(file_storage_dir):
227
    raise errors.OpPrereqError("File storage directory '%s' is not an absolute"
228
                               " path" % file_storage_dir, errors.ECODE_INVAL)
229

    
230
  if not os.path.exists(file_storage_dir):
231
    try:
232
      os.makedirs(file_storage_dir, 0750)
233
    except OSError, err:
234
      raise errors.OpPrereqError("Cannot create file storage directory"
235
                                 " '%s': %s" % (file_storage_dir, err),
236
                                 errors.ECODE_ENVIRON)
237

    
238
  if not os.path.isdir(file_storage_dir):
239
    raise errors.OpPrereqError("The file storage directory '%s' is not"
240
                               " a directory." % file_storage_dir,
241
                               errors.ECODE_ENVIRON)
242
  return file_storage_dir
243

    
244

    
245
def InitCluster(cluster_name, mac_prefix, # pylint: disable-msg=R0913
246
                master_netdev, file_storage_dir, shared_file_storage_dir,
247
                candidate_pool_size, secondary_ip=None, vg_name=None,
248
                beparams=None, nicparams=None, ndparams=None, hvparams=None,
249
                enabled_hypervisors=None, modify_etc_hosts=True,
250
                modify_ssh_setup=True, maintain_node_health=False,
251
                drbd_helper=None, uid_pool=None, default_iallocator=None,
252
                primary_ip_version=None, prealloc_wipe_disks=False):
253
  """Initialise the cluster.
254

255
  @type candidate_pool_size: int
256
  @param candidate_pool_size: master candidate pool size
257

258
  """
259
  # TODO: complete the docstring
260
  if config.ConfigWriter.IsCluster():
261
    raise errors.OpPrereqError("Cluster is already initialised",
262
                               errors.ECODE_STATE)
263

    
264
  if not enabled_hypervisors:
265
    raise errors.OpPrereqError("Enabled hypervisors list must contain at"
266
                               " least one member", errors.ECODE_INVAL)
267
  invalid_hvs = set(enabled_hypervisors) - constants.HYPER_TYPES
268
  if invalid_hvs:
269
    raise errors.OpPrereqError("Enabled hypervisors contains invalid"
270
                               " entries: %s" % invalid_hvs,
271
                               errors.ECODE_INVAL)
272

    
273

    
274
  ipcls = None
275
  if primary_ip_version == constants.IP4_VERSION:
276
    ipcls = netutils.IP4Address
277
  elif primary_ip_version == constants.IP6_VERSION:
278
    ipcls = netutils.IP6Address
279
  else:
280
    raise errors.OpPrereqError("Invalid primary ip version: %d." %
281
                               primary_ip_version)
282

    
283
  hostname = netutils.GetHostname(family=ipcls.family)
284
  if not ipcls.IsValid(hostname.ip):
285
    raise errors.OpPrereqError("This host's IP (%s) is not a valid IPv%d"
286
                               " address." % (hostname.ip, primary_ip_version))
287

    
288
  if ipcls.IsLoopback(hostname.ip):
289
    raise errors.OpPrereqError("This host's IP (%s) resolves to a loopback"
290
                               " address. Please fix DNS or %s." %
291
                               (hostname.ip, constants.ETC_HOSTS),
292
                               errors.ECODE_ENVIRON)
293

    
294
  if not ipcls.Own(hostname.ip):
295
    raise errors.OpPrereqError("Inconsistency: this host's name resolves"
296
                               " to %s,\nbut this ip address does not"
297
                               " belong to this host" %
298
                               hostname.ip, errors.ECODE_ENVIRON)
299

    
300
  clustername = netutils.GetHostname(name=cluster_name, family=ipcls.family)
301

    
302
  if netutils.TcpPing(clustername.ip, constants.DEFAULT_NODED_PORT, timeout=5):
303
    raise errors.OpPrereqError("Cluster IP already active",
304
                               errors.ECODE_NOTUNIQUE)
305

    
306
  if not secondary_ip:
307
    if primary_ip_version == constants.IP6_VERSION:
308
      raise errors.OpPrereqError("When using a IPv6 primary address, a valid"
309
                                 " IPv4 address must be given as secondary",
310
                                 errors.ECODE_INVAL)
311
    secondary_ip = hostname.ip
312

    
313
  if not netutils.IP4Address.IsValid(secondary_ip):
314
    raise errors.OpPrereqError("Secondary IP address (%s) has to be a valid"
315
                               " IPv4 address." % secondary_ip,
316
                               errors.ECODE_INVAL)
317

    
318
  if not netutils.IP4Address.Own(secondary_ip):
319
    raise errors.OpPrereqError("You gave %s as secondary IP,"
320
                               " but it does not belong to this host." %
321
                               secondary_ip, errors.ECODE_ENVIRON)
322

    
323
  if vg_name is not None:
324
    # Check if volume group is valid
325
    vgstatus = utils.CheckVolumeGroupSize(utils.ListVolumeGroups(), vg_name,
326
                                          constants.MIN_VG_SIZE)
327
    if vgstatus:
328
      raise errors.OpPrereqError("Error: %s\nspecify --no-lvm-storage if"
329
                                 " you are not using lvm" % vgstatus,
330
                                 errors.ECODE_INVAL)
331

    
332
  if drbd_helper is not None:
333
    try:
334
      curr_helper = bdev.BaseDRBD.GetUsermodeHelper()
335
    except errors.BlockDeviceError, err:
336
      raise errors.OpPrereqError("Error while checking drbd helper"
337
                                 " (specify --no-drbd-storage if you are not"
338
                                 " using drbd): %s" % str(err),
339
                                 errors.ECODE_ENVIRON)
340
    if drbd_helper != curr_helper:
341
      raise errors.OpPrereqError("Error: requiring %s as drbd helper but %s"
342
                                 " is the current helper" % (drbd_helper,
343
                                                             curr_helper),
344
                                 errors.ECODE_INVAL)
345

    
346
  if constants.ENABLE_FILE_STORAGE:
347
    file_storage_dir = _InitFileStorage(file_storage_dir)
348
  else:
349
    file_storage_dir = ""
350

    
351
  if constants.ENABLE_SHARED_FILE_STORAGE:
352
    shared_file_storage_dir = _InitFileStorage(shared_file_storage_dir)
353
  else:
354
    shared_file_storage_dir = ""
355

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

    
360
  result = utils.RunCmd(["ip", "link", "show", "dev", master_netdev])
361
  if result.failed:
362
    raise errors.OpPrereqError("Invalid master netdev given (%s): '%s'" %
363
                               (master_netdev,
364
                                result.output.strip()), errors.ECODE_INVAL)
365

    
366
  dirs = [(constants.RUN_GANETI_DIR, constants.RUN_DIRS_MODE)]
367
  utils.EnsureDirs(dirs)
368

    
369
  utils.ForceDictType(beparams, constants.BES_PARAMETER_TYPES)
370
  utils.ForceDictType(nicparams, constants.NICS_PARAMETER_TYPES)
371
  objects.NIC.CheckParameterSyntax(nicparams)
372

    
373
  if ndparams is not None:
374
    utils.ForceDictType(ndparams, constants.NDS_PARAMETER_TYPES)
375
  else:
376
    ndparams = dict(constants.NDC_DEFAULTS)
377

    
378
  # hvparams is a mapping of hypervisor->hvparams dict
379
  for hv_name, hv_params in hvparams.iteritems():
380
    utils.ForceDictType(hv_params, constants.HVS_PARAMETER_TYPES)
381
    hv_class = hypervisor.GetHypervisor(hv_name)
382
    hv_class.CheckParameterSyntax(hv_params)
383

    
384
  # set up ssh config and /etc/hosts
385
  sshline = utils.ReadFile(constants.SSH_HOST_RSA_PUB)
386
  sshkey = sshline.split(" ")[1]
387

    
388
  if modify_etc_hosts:
389
    utils.AddHostToEtcHosts(hostname.name, hostname.ip)
390

    
391
  if modify_ssh_setup:
392
    _InitSSHSetup()
393

    
394
  if default_iallocator is not None:
395
    alloc_script = utils.FindFile(default_iallocator,
396
                                  constants.IALLOCATOR_SEARCH_PATH,
397
                                  os.path.isfile)
398
    if alloc_script is None:
399
      raise errors.OpPrereqError("Invalid default iallocator script '%s'"
400
                                 " specified" % default_iallocator,
401
                                 errors.ECODE_INVAL)
402
  elif constants.HTOOLS:
403
    # htools was enabled at build-time, we default to it
404
    if utils.FindFile(constants.IALLOC_HAIL,
405
                      constants.IALLOCATOR_SEARCH_PATH,
406
                      os.path.isfile):
407
      default_iallocator = constants.IALLOC_HAIL
408

    
409
  now = time.time()
410

    
411
  # init of cluster config file
412
  cluster_config = objects.Cluster(
413
    serial_no=1,
414
    rsahostkeypub=sshkey,
415
    highest_used_port=(constants.FIRST_DRBD_PORT - 1),
416
    mac_prefix=mac_prefix,
417
    volume_group_name=vg_name,
418
    tcpudp_port_pool=set(),
419
    master_node=hostname.name,
420
    master_ip=clustername.ip,
421
    master_netdev=master_netdev,
422
    cluster_name=clustername.name,
423
    file_storage_dir=file_storage_dir,
424
    shared_file_storage_dir=shared_file_storage_dir,
425
    enabled_hypervisors=enabled_hypervisors,
426
    beparams={constants.PP_DEFAULT: beparams},
427
    nicparams={constants.PP_DEFAULT: nicparams},
428
    ndparams=ndparams,
429
    hvparams=hvparams,
430
    candidate_pool_size=candidate_pool_size,
431
    modify_etc_hosts=modify_etc_hosts,
432
    modify_ssh_setup=modify_ssh_setup,
433
    uid_pool=uid_pool,
434
    ctime=now,
435
    mtime=now,
436
    maintain_node_health=maintain_node_health,
437
    drbd_usermode_helper=drbd_helper,
438
    default_iallocator=default_iallocator,
439
    primary_ip_family=ipcls.family,
440
    prealloc_wipe_disks=prealloc_wipe_disks,
441
    )
442
  master_node_config = objects.Node(name=hostname.name,
443
                                    primary_ip=hostname.ip,
444
                                    secondary_ip=secondary_ip,
445
                                    serial_no=1,
446
                                    master_candidate=True,
447
                                    offline=False, drained=False,
448
                                    ctime=now, mtime=now,
449
                                    )
450
  InitConfig(constants.CONFIG_VERSION, cluster_config, master_node_config)
451
  cfg = config.ConfigWriter(offline=True)
452
  ssh.WriteKnownHostsFile(cfg, constants.SSH_KNOWN_HOSTS_FILE)
453
  cfg.Update(cfg.GetClusterInfo(), logging.error)
454
  backend.WriteSsconfFiles(cfg.GetSsconfValues())
455

    
456
  # set up the inter-node password and certificate
457
  _InitGanetiServerSetup(hostname.name)
458

    
459
  logging.debug("Starting daemons")
460
  result = utils.RunCmd([constants.DAEMON_UTIL, "start-all"])
461
  if result.failed:
462
    raise errors.OpExecError("Could not start daemons, command %s"
463
                             " had exitcode %s and error %s" %
464
                             (result.cmd, result.exit_code, result.output))
465

    
466
  _WaitForMasterDaemon()
467

    
468

    
469
def InitConfig(version, cluster_config, master_node_config,
470
               cfg_file=constants.CLUSTER_CONF_FILE):
471
  """Create the initial cluster configuration.
472

473
  It will contain the current node, which will also be the master
474
  node, and no instances.
475

476
  @type version: int
477
  @param version: configuration version
478
  @type cluster_config: L{objects.Cluster}
479
  @param cluster_config: cluster configuration
480
  @type master_node_config: L{objects.Node}
481
  @param master_node_config: master node configuration
482
  @type cfg_file: string
483
  @param cfg_file: configuration file path
484

485
  """
486
  uuid_generator = config.TemporaryReservationManager()
487
  cluster_config.uuid = uuid_generator.Generate([], utils.NewUUID,
488
                                                _INITCONF_ECID)
489
  master_node_config.uuid = uuid_generator.Generate([], utils.NewUUID,
490
                                                    _INITCONF_ECID)
491
  nodes = {
492
    master_node_config.name: master_node_config,
493
    }
494
  default_nodegroup = objects.NodeGroup(
495
    uuid=uuid_generator.Generate([], utils.NewUUID, _INITCONF_ECID),
496
    name=constants.INITIAL_NODE_GROUP_NAME,
497
    members=[master_node_config.name],
498
    )
499
  nodegroups = {
500
    default_nodegroup.uuid: default_nodegroup,
501
    }
502
  now = time.time()
503
  config_data = objects.ConfigData(version=version,
504
                                   cluster=cluster_config,
505
                                   nodegroups=nodegroups,
506
                                   nodes=nodes,
507
                                   instances={},
508
                                   serial_no=1,
509
                                   ctime=now, mtime=now)
510
  utils.WriteFile(cfg_file,
511
                  data=serializer.Dump(config_data.ToDict()),
512
                  mode=0600)
513

    
514

    
515
def FinalizeClusterDestroy(master):
516
  """Execute the last steps of cluster destroy
517

518
  This function shuts down all the daemons, completing the destroy
519
  begun in cmdlib.LUDestroyOpcode.
520

521
  """
522
  cfg = config.ConfigWriter()
523
  modify_ssh_setup = cfg.GetClusterInfo().modify_ssh_setup
524
  result = rpc.RpcRunner.call_node_stop_master(master, True)
525
  msg = result.fail_msg
526
  if msg:
527
    logging.warning("Could not disable the master role: %s", msg)
528
  result = rpc.RpcRunner.call_node_leave_cluster(master, modify_ssh_setup)
529
  msg = result.fail_msg
530
  if msg:
531
    logging.warning("Could not shutdown the node daemon and cleanup"
532
                    " the node: %s", msg)
533

    
534

    
535
def SetupNodeDaemon(cluster_name, node, ssh_key_check):
536
  """Add a node to the cluster.
537

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

542
  @param cluster_name: the cluster name
543
  @param node: the name of the new node
544
  @param ssh_key_check: whether to do a strict key check
545

546
  """
547
  family = ssconf.SimpleStore().GetPrimaryIPFamily()
548
  sshrunner = ssh.SshRunner(cluster_name,
549
                            ipv6=(family == netutils.IP6Address.family))
550

    
551
  bind_address = constants.IP4_ADDRESS_ANY
552
  if family == netutils.IP6Address.family:
553
    bind_address = constants.IP6_ADDRESS_ANY
554

    
555
  # set up inter-node password and certificate and restarts the node daemon
556
  # and then connect with ssh to set password and start ganeti-noded
557
  # note that all the below variables are sanitized at this point,
558
  # either by being constants or by the checks above
559
  sshrunner.CopyFileToNode(node, constants.NODED_CERT_FILE)
560
  sshrunner.CopyFileToNode(node, constants.RAPI_CERT_FILE)
561
  sshrunner.CopyFileToNode(node, constants.CONFD_HMAC_KEY)
562
  mycommand = ("%s stop-all; %s start %s -b %s" %
563
               (constants.DAEMON_UTIL, constants.DAEMON_UTIL, constants.NODED,
564
                utils.ShellQuote(bind_address)))
565

    
566
  result = sshrunner.Run(node, 'root', mycommand, batch=False,
567
                         ask_key=ssh_key_check,
568
                         use_cluster_key=True,
569
                         strict_host_check=ssh_key_check)
570
  if result.failed:
571
    raise errors.OpExecError("Remote command on node %s, error: %s,"
572
                             " output: %s" %
573
                             (node, result.fail_reason, result.output))
574

    
575
  _WaitForNodeDaemon(node)
576

    
577

    
578
def MasterFailover(no_voting=False):
579
  """Failover the master node.
580

581
  This checks that we are not already the master, and will cause the
582
  current master to cease being master, and the non-master to become
583
  new master.
584

585
  @type no_voting: boolean
586
  @param no_voting: force the operation without remote nodes agreement
587
                      (dangerous)
588

589
  """
590
  sstore = ssconf.SimpleStore()
591

    
592
  old_master, new_master = ssconf.GetMasterAndMyself(sstore)
593
  node_list = sstore.GetNodeList()
594
  mc_list = sstore.GetMasterCandidates()
595

    
596
  if old_master == new_master:
597
    raise errors.OpPrereqError("This commands must be run on the node"
598
                               " where you want the new master to be."
599
                               " %s is already the master" %
600
                               old_master, errors.ECODE_INVAL)
601

    
602
  if new_master not in mc_list:
603
    mc_no_master = [name for name in mc_list if name != old_master]
604
    raise errors.OpPrereqError("This node is not among the nodes marked"
605
                               " as master candidates. Only these nodes"
606
                               " can become masters. Current list of"
607
                               " master candidates is:\n"
608
                               "%s" % ('\n'.join(mc_no_master)),
609
                               errors.ECODE_STATE)
610

    
611
  if not no_voting:
612
    vote_list = GatherMasterVotes(node_list)
613

    
614
    if vote_list:
615
      voted_master = vote_list[0][0]
616
      if voted_master is None:
617
        raise errors.OpPrereqError("Cluster is inconsistent, most nodes did"
618
                                   " not respond.", errors.ECODE_ENVIRON)
619
      elif voted_master != old_master:
620
        raise errors.OpPrereqError("I have a wrong configuration, I believe"
621
                                   " the master is %s but the other nodes"
622
                                   " voted %s. Please resync the configuration"
623
                                   " of this node." %
624
                                   (old_master, voted_master),
625
                                   errors.ECODE_STATE)
626
  # end checks
627

    
628
  rcode = 0
629

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

    
632
  try:
633
    # instantiate a real config writer, as we now know we have the
634
    # configuration data
635
    cfg = config.ConfigWriter(accept_foreign=True)
636

    
637
    cluster_info = cfg.GetClusterInfo()
638
    cluster_info.master_node = new_master
639
    # this will also regenerate the ssconf files, since we updated the
640
    # cluster info
641
    cfg.Update(cluster_info, logging.error)
642
  except errors.ConfigurationError, err:
643
    logging.error("Error while trying to set the new master: %s",
644
                  str(err))
645
    return 1
646

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

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

    
654
  result = rpc.RpcRunner.call_node_stop_master(old_master, True)
655
  msg = result.fail_msg
656
  if msg:
657
    logging.error("Could not disable the master role on the old master"
658
                 " %s, please disable manually: %s", old_master, msg)
659

    
660
  logging.info("Checking master IP non-reachability...")
661

    
662
  master_ip = sstore.GetMasterIP()
663
  total_timeout = 30
664
  # Here we have a phase where no master should be running
665
  def _check_ip():
666
    if netutils.TcpPing(master_ip, constants.DEFAULT_NODED_PORT):
667
      raise utils.RetryAgain()
668

    
669
  try:
670
    utils.Retry(_check_ip, (1, 1.5, 5), total_timeout)
671
  except utils.RetryTimeout:
672
    logging.warning("The master IP is still reachable after %s seconds,"
673
                    " continuing but activating the master on the current"
674
                    " node will probably fail", total_timeout)
675

    
676
  if jstore.CheckDrainFlag():
677
    logging.info("Undraining job queue")
678
    jstore.SetDrainFlag(False)
679

    
680
  logging.info("Starting the master daemons on the new master")
681

    
682
  result = rpc.RpcRunner.call_node_start_master(new_master, True, no_voting)
683
  msg = result.fail_msg
684
  if msg:
685
    logging.error("Could not start the master role on the new master"
686
                  " %s, please check: %s", new_master, msg)
687
    rcode = 1
688

    
689
  logging.info("Master failed over from %s to %s", old_master, new_master)
690
  return rcode
691

    
692

    
693
def GetMaster():
694
  """Returns the current master node.
695

696
  This is a separate function in bootstrap since it's needed by
697
  gnt-cluster, and instead of importing directly ssconf, it's better
698
  to abstract it in bootstrap, where we do use ssconf in other
699
  functions too.
700

701
  """
702
  sstore = ssconf.SimpleStore()
703

    
704
  old_master, _ = ssconf.GetMasterAndMyself(sstore)
705

    
706
  return old_master
707

    
708

    
709
def GatherMasterVotes(node_list):
710
  """Check the agreement on who is the master.
711

712
  This function will return a list of (node, number of votes), ordered
713
  by the number of votes. Errors will be denoted by the key 'None'.
714

715
  Note that the sum of votes is the number of nodes this machine
716
  knows, whereas the number of entries in the list could be different
717
  (if some nodes vote for another master).
718

719
  We remove ourselves from the list since we know that (bugs aside)
720
  since we use the same source for configuration information for both
721
  backend and boostrap, we'll always vote for ourselves.
722

723
  @type node_list: list
724
  @param node_list: the list of nodes to query for master info; the current
725
      node will be removed if it is in the list
726
  @rtype: list
727
  @return: list of (node, votes)
728

729
  """
730
  myself = netutils.Hostname.GetSysName()
731
  try:
732
    node_list.remove(myself)
733
  except ValueError:
734
    pass
735
  if not node_list:
736
    # no nodes left (eventually after removing myself)
737
    return []
738
  results = rpc.RpcRunner.call_master_info(node_list)
739
  if not isinstance(results, dict):
740
    # this should not happen (unless internal error in rpc)
741
    logging.critical("Can't complete rpc call, aborting master startup")
742
    return [(None, len(node_list))]
743
  votes = {}
744
  for node in results:
745
    nres = results[node]
746
    data = nres.payload
747
    msg = nres.fail_msg
748
    fail = False
749
    if msg:
750
      logging.warning("Error contacting node %s: %s", node, msg)
751
      fail = True
752
    # for now we accept both length 3 and 4 (data[3] is primary ip version)
753
    elif not isinstance(data, (tuple, list)) or len(data) < 3:
754
      logging.warning("Invalid data received from node %s: %s", node, data)
755
      fail = True
756
    if fail:
757
      if None not in votes:
758
        votes[None] = 0
759
      votes[None] += 1
760
      continue
761
    master_node = data[2]
762
    if master_node not in votes:
763
      votes[master_node] = 0
764
    votes[master_node] += 1
765

    
766
  vote_list = [v for v in votes.items()]
767
  # sort first on number of votes then on name, since we want None
768
  # sorted later if we have the half of the nodes not responding, and
769
  # half voting all for the same master
770
  vote_list.sort(key=lambda x: (x[1], x[0]), reverse=True)
771

    
772
  return vote_list