code.grnet.gr Git - ganeti-local/blob - lib/bootstrap.py

   1 #
   2 #
   3
   4 # Copyright (C) 2006, 2007, 2008 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
  43
  44 def _InitSSHSetup():
  45   """Setup the SSH configuration for the cluster.
  46
  47   This generates a dsa keypair for root, adds the pub key to the
  48   permitted hosts and adds the hostkey to its own known hosts.
  49
  50   """
  51   priv_key, pub_key, auth_keys = ssh.GetUserFiles(constants.GANETI_RUNAS)
  52
  53   for name in priv_key, pub_key:
  54     if os.path.exists(name):
  55       utils.CreateBackup(name)
  56     utils.RemoveFile(name)
  57
  58   result = utils.RunCmd(["ssh-keygen", "-t", "dsa",
  59                          "-f", priv_key,
  60                          "-q", "-N", ""])
  61   if result.failed:
  62     raise errors.OpExecError("Could not generate ssh keypair, error %s" %
  63                              result.output)
  64
  65   utils.AddAuthorizedKey(auth_keys, utils.ReadFile(pub_key))
  66
  67
  68 def GenerateHmacKey(file_name):
  69   """Writes a new HMAC key.
  70
  71   @type file_name: str
  72   @param file_name: Path to output file
  73
  74   """
  75   utils.WriteFile(file_name, data="%s\n" % utils.GenerateSecret(), mode=0400,
  76                   backup=True)
  77
  78
  79 def GenerateClusterCrypto(new_cluster_cert, new_rapi_cert, new_confd_hmac_key,
  80                           new_cds, rapi_cert_pem=None, cds=None):
  81   """Updates the cluster certificates, keys and secrets.
  82
  83   @type new_cluster_cert: bool
  84   @param new_cluster_cert: Whether to generate a new cluster certificate
  85   @type new_rapi_cert: bool
  86   @param new_rapi_cert: Whether to generate a new RAPI certificate
  87   @type new_confd_hmac_key: bool
  88   @param new_confd_hmac_key: Whether to generate a new HMAC key
  89   @type new_cds: bool
  90   @param new_cds: Whether to generate a new cluster domain secret
  91   @type rapi_cert_pem: string
  92   @param rapi_cert_pem: New RAPI certificate in PEM format
  93   @type cds: string
  94   @param cds: New cluster domain secret
  95
  96   """
  97   # noded SSL certificate
  98   cluster_cert_exists = os.path.exists(constants.NODED_CERT_FILE)
  99   if new_cluster_cert or not cluster_cert_exists:
 100     if cluster_cert_exists:
 101       utils.CreateBackup(constants.NODED_CERT_FILE)
 102
 103     logging.debug("Generating new cluster certificate at %s",
 104                   constants.NODED_CERT_FILE)
 105     utils.GenerateSelfSignedSslCert(constants.NODED_CERT_FILE)
 106
 107   # confd HMAC key
 108   if new_confd_hmac_key or not os.path.exists(constants.CONFD_HMAC_KEY):
 109     logging.debug("Writing new confd HMAC key to %s", constants.CONFD_HMAC_KEY)
 110     GenerateHmacKey(constants.CONFD_HMAC_KEY)
 111
 112   # RAPI
 113   rapi_cert_exists = os.path.exists(constants.RAPI_CERT_FILE)
 114
 115   if rapi_cert_pem:
 116     # Assume rapi_pem contains a valid PEM-formatted certificate and key
 117     logging.debug("Writing RAPI certificate at %s",
 118                   constants.RAPI_CERT_FILE)
 119     utils.WriteFile(constants.RAPI_CERT_FILE, data=rapi_cert_pem, backup=True)
 120
 121   elif new_rapi_cert or not rapi_cert_exists:
 122     if rapi_cert_exists:
 123       utils.CreateBackup(constants.RAPI_CERT_FILE)
 124
 125     logging.debug("Generating new RAPI certificate at %s",
 126                   constants.RAPI_CERT_FILE)
 127     utils.GenerateSelfSignedSslCert(constants.RAPI_CERT_FILE)
 128
 129   # Cluster domain secret
 130   if cds:
 131     logging.debug("Writing cluster domain secret to %s",
 132                   constants.CLUSTER_DOMAIN_SECRET_FILE)
 133     utils.WriteFile(constants.CLUSTER_DOMAIN_SECRET_FILE,
 134                     data=cds, backup=True)
 135
 136   elif new_cds or not os.path.exists(constants.CLUSTER_DOMAIN_SECRET_FILE):
 137     logging.debug("Generating new cluster domain secret at %s",
 138                   constants.CLUSTER_DOMAIN_SECRET_FILE)
 139     GenerateHmacKey(constants.CLUSTER_DOMAIN_SECRET_FILE)
 140
 141
 142 def _InitGanetiServerSetup(master_name):
 143   """Setup the necessary configuration for the initial node daemon.
 144
 145   This creates the nodepass file containing the shared password for
 146   the cluster and also generates the SSL certificate.
 147
 148   """
 149   # Generate cluster secrets
 150   GenerateClusterCrypto(True, False, False, False)
 151
 152   result = utils.RunCmd([constants.DAEMON_UTIL, "start", constants.NODED])
 153   if result.failed:
 154     raise errors.OpExecError("Could not start the node daemon, command %s"
 155                              " had exitcode %s and error %s" %
 156                              (result.cmd, result.exit_code, result.output))
 157
 158   _WaitForNodeDaemon(master_name)
 159
 160
 161 def _WaitForNodeDaemon(node_name):
 162   """Wait for node daemon to become responsive.
 163
 164   """
 165   def _CheckNodeDaemon():
 166     result = rpc.RpcRunner.call_version([node_name])[node_name]
 167     if result.fail_msg:
 168       raise utils.RetryAgain()
 169
 170   try:
 171     utils.Retry(_CheckNodeDaemon, 1.0, 10.0)
 172   except utils.RetryTimeout:
 173     raise errors.OpExecError("Node daemon on %s didn't answer queries within"
 174                              " 10 seconds" % node_name)
 175
 176
 177 def InitCluster(cluster_name, mac_prefix,
 178                 master_netdev, file_storage_dir, candidate_pool_size,
 179                 secondary_ip=None, vg_name=None, beparams=None,
 180                 nicparams=None, hvparams=None, enabled_hypervisors=None,
 181                 modify_etc_hosts=True, modify_ssh_setup=True):
 182   """Initialise the cluster.
 183
 184   @type candidate_pool_size: int
 185   @param candidate_pool_size: master candidate pool size
 186
 187   """
 188   # TODO: complete the docstring
 189   if config.ConfigWriter.IsCluster():
 190     raise errors.OpPrereqError("Cluster is already initialised",
 191                                errors.ECODE_STATE)
 192
 193   if not enabled_hypervisors:
 194     raise errors.OpPrereqError("Enabled hypervisors list must contain at"
 195                                " least one member", errors.ECODE_INVAL)
 196   invalid_hvs = set(enabled_hypervisors) - constants.HYPER_TYPES
 197   if invalid_hvs:
 198     raise errors.OpPrereqError("Enabled hypervisors contains invalid"
 199                                " entries: %s" % invalid_hvs,
 200                                errors.ECODE_INVAL)
 201
 202   hostname = utils.GetHostInfo()
 203
 204   if hostname.ip.startswith("127."):
 205     raise errors.OpPrereqError("This host's IP resolves to the private"
 206                                " range (%s). Please fix DNS or %s." %
 207                                (hostname.ip, constants.ETC_HOSTS),
 208                                errors.ECODE_ENVIRON)
 209
 210   if not utils.OwnIpAddress(hostname.ip):
 211     raise errors.OpPrereqError("Inconsistency: this host's name resolves"
 212                                " to %s,\nbut this ip address does not"
 213                                " belong to this host. Aborting." %
 214                                hostname.ip, errors.ECODE_ENVIRON)
 215
 216   clustername = utils.GetHostInfo(utils.HostInfo.NormalizeName(cluster_name))
 217
 218   if utils.TcpPing(clustername.ip, constants.DEFAULT_NODED_PORT,
 219                    timeout=5):
 220     raise errors.OpPrereqError("Cluster IP already active. Aborting.",
 221                                errors.ECODE_NOTUNIQUE)
 222
 223   if secondary_ip:
 224     if not utils.IsValidIP(secondary_ip):
 225       raise errors.OpPrereqError("Invalid secondary ip given",
 226                                  errors.ECODE_INVAL)
 227     if (secondary_ip != hostname.ip and
 228         not utils.OwnIpAddress(secondary_ip)):
 229       raise errors.OpPrereqError("You gave %s as secondary IP,"
 230                                  " but it does not belong to this host." %
 231                                  secondary_ip, errors.ECODE_ENVIRON)
 232   else:
 233     secondary_ip = hostname.ip
 234
 235   if vg_name is not None:
 236     # Check if volume group is valid
 237     vgstatus = utils.CheckVolumeGroupSize(utils.ListVolumeGroups(), vg_name,
 238                                           constants.MIN_VG_SIZE)
 239     if vgstatus:
 240       raise errors.OpPrereqError("Error: %s\nspecify --no-lvm-storage if"
 241                                  " you are not using lvm" % vgstatus,
 242                                  errors.ECODE_INVAL)
 243
 244   file_storage_dir = os.path.normpath(file_storage_dir)
 245
 246   if not os.path.isabs(file_storage_dir):
 247     raise errors.OpPrereqError("The file storage directory you passed is"
 248                                " not an absolute path.", errors.ECODE_INVAL)
 249
 250   if not os.path.exists(file_storage_dir):
 251     try:
 252       os.makedirs(file_storage_dir, 0750)
 253     except OSError, err:
 254       raise errors.OpPrereqError("Cannot create file storage directory"
 255                                  " '%s': %s" % (file_storage_dir, err),
 256                                  errors.ECODE_ENVIRON)
 257
 258   if not os.path.isdir(file_storage_dir):
 259     raise errors.OpPrereqError("The file storage directory '%s' is not"
 260                                " a directory." % file_storage_dir,
 261                                errors.ECODE_ENVIRON)
 262
 263   if not re.match("^[0-9a-z]{2}:[0-9a-z]{2}:[0-9a-z]{2}$", mac_prefix):
 264     raise errors.OpPrereqError("Invalid mac prefix given '%s'" % mac_prefix,
 265                                errors.ECODE_INVAL)
 266
 267   result = utils.RunCmd(["ip", "link", "show", "dev", master_netdev])
 268   if result.failed:
 269     raise errors.OpPrereqError("Invalid master netdev given (%s): '%s'" %
 270                                (master_netdev,
 271                                 result.output.strip()), errors.ECODE_INVAL)
 272
 273   dirs = [(constants.RUN_GANETI_DIR, constants.RUN_DIRS_MODE)]
 274   utils.EnsureDirs(dirs)
 275
 276   utils.ForceDictType(beparams, constants.BES_PARAMETER_TYPES)
 277   utils.ForceDictType(nicparams, constants.NICS_PARAMETER_TYPES)
 278   objects.NIC.CheckParameterSyntax(nicparams)
 279
 280   # hvparams is a mapping of hypervisor->hvparams dict
 281   for hv_name, hv_params in hvparams.iteritems():
 282     utils.ForceDictType(hv_params, constants.HVS_PARAMETER_TYPES)
 283     hv_class = hypervisor.GetHypervisor(hv_name)
 284     hv_class.CheckParameterSyntax(hv_params)
 285
 286   # set up the inter-node password and certificate
 287   _InitGanetiServerSetup(hostname.name)
 288
 289   # set up ssh config and /etc/hosts
 290   sshline = utils.ReadFile(constants.SSH_HOST_RSA_PUB)
 291   sshkey = sshline.split(" ")[1]
 292
 293   if modify_etc_hosts:
 294     utils.AddHostToEtcHosts(hostname.name)
 295
 296   if modify_ssh_setup:
 297     _InitSSHSetup()
 298
 299   now = time.time()
 300
 301   # init of cluster config file
 302   cluster_config = objects.Cluster(
 303     serial_no=1,
 304     rsahostkeypub=sshkey,
 305     highest_used_port=(constants.FIRST_DRBD_PORT - 1),
 306     mac_prefix=mac_prefix,
 307     volume_group_name=vg_name,
 308     tcpudp_port_pool=set(),
 309     master_node=hostname.name,
 310     master_ip=clustername.ip,
 311     master_netdev=master_netdev,
 312     cluster_name=clustername.name,
 313     file_storage_dir=file_storage_dir,
 314     enabled_hypervisors=enabled_hypervisors,
 315     beparams={constants.PP_DEFAULT: beparams},
 316     nicparams={constants.PP_DEFAULT: nicparams},
 317     hvparams=hvparams,
 318     candidate_pool_size=candidate_pool_size,
 319     modify_etc_hosts=modify_etc_hosts,
 320     modify_ssh_setup=modify_ssh_setup,
 321     ctime=now,
 322     mtime=now,
 323     uuid=utils.NewUUID(),
 324     )
 325   master_node_config = objects.Node(name=hostname.name,
 326                                     primary_ip=hostname.ip,
 327                                     secondary_ip=secondary_ip,
 328                                     serial_no=1,
 329                                     master_candidate=True,
 330                                     offline=False, drained=False,
 331                                     )
 332   InitConfig(constants.CONFIG_VERSION, cluster_config, master_node_config)
 333   cfg = config.ConfigWriter()
 334   ssh.WriteKnownHostsFile(cfg, constants.SSH_KNOWN_HOSTS_FILE)
 335   cfg.Update(cfg.GetClusterInfo(), logging.error)
 336
 337   # start the master ip
 338   # TODO: Review rpc call from bootstrap
 339   # TODO: Warn on failed start master
 340   rpc.RpcRunner.call_node_start_master(hostname.name, True, False)
 341
 342
 343 def InitConfig(version, cluster_config, master_node_config,
 344                cfg_file=constants.CLUSTER_CONF_FILE):
 345   """Create the initial cluster configuration.
 346
 347   It will contain the current node, which will also be the master
 348   node, and no instances.
 349
 350   @type version: int
 351   @param version: configuration version
 352   @type cluster_config: L{objects.Cluster}
 353   @param cluster_config: cluster configuration
 354   @type master_node_config: L{objects.Node}
 355   @param master_node_config: master node configuration
 356   @type cfg_file: string
 357   @param cfg_file: configuration file path
 358
 359   """
 360   nodes = {
 361     master_node_config.name: master_node_config,
 362     }
 363
 364   now = time.time()
 365   config_data = objects.ConfigData(version=version,
 366                                    cluster=cluster_config,
 367                                    nodes=nodes,
 368                                    instances={},
 369                                    serial_no=1,
 370                                    ctime=now, mtime=now)
 371   utils.WriteFile(cfg_file,
 372                   data=serializer.Dump(config_data.ToDict()),
 373                   mode=0600)
 374
 375
 376 def FinalizeClusterDestroy(master):
 377   """Execute the last steps of cluster destroy
 378
 379   This function shuts down all the daemons, completing the destroy
 380   begun in cmdlib.LUDestroyOpcode.
 381
 382   """
 383   cfg = config.ConfigWriter()
 384   modify_ssh_setup = cfg.GetClusterInfo().modify_ssh_setup
 385   result = rpc.RpcRunner.call_node_stop_master(master, True)
 386   msg = result.fail_msg
 387   if msg:
 388     logging.warning("Could not disable the master role: %s", msg)
 389   result = rpc.RpcRunner.call_node_leave_cluster(master, modify_ssh_setup)
 390   msg = result.fail_msg
 391   if msg:
 392     logging.warning("Could not shutdown the node daemon and cleanup"
 393                     " the node: %s", msg)
 394
 395
 396 def SetupNodeDaemon(cluster_name, node, ssh_key_check):
 397   """Add a node to the cluster.
 398
 399   This function must be called before the actual opcode, and will ssh
 400   to the remote node, copy the needed files, and start ganeti-noded,
 401   allowing the master to do the rest via normal rpc calls.
 402
 403   @param cluster_name: the cluster name
 404   @param node: the name of the new node
 405   @param ssh_key_check: whether to do a strict key check
 406
 407   """
 408   sshrunner = ssh.SshRunner(cluster_name)
 409
 410   noded_cert = utils.ReadFile(constants.NODED_CERT_FILE)
 411   rapi_cert = utils.ReadFile(constants.RAPI_CERT_FILE)
 412   confd_hmac_key = utils.ReadFile(constants.CONFD_HMAC_KEY)
 413
 414   # in the base64 pem encoding, neither '!' nor '.' are valid chars,
 415   # so we use this to detect an invalid certificate; as long as the
 416   # cert doesn't contain this, the here-document will be correctly
 417   # parsed by the shell sequence below. HMAC keys are hexadecimal strings,
 418   # so the same restrictions apply.
 419   for content in (noded_cert, rapi_cert, confd_hmac_key):
 420     if re.search('^!EOF\.', content, re.MULTILINE):
 421       raise errors.OpExecError("invalid SSL certificate or HMAC key")
 422
 423   if not noded_cert.endswith("\n"):
 424     noded_cert += "\n"
 425   if not rapi_cert.endswith("\n"):
 426     rapi_cert += "\n"
 427   if not confd_hmac_key.endswith("\n"):
 428     confd_hmac_key += "\n"
 429
 430   # set up inter-node password and certificate and restarts the node daemon
 431   # and then connect with ssh to set password and start ganeti-noded
 432   # note that all the below variables are sanitized at this point,
 433   # either by being constants or by the checks above
 434   # TODO: Could this command exceed a shell's maximum command length?
 435   mycommand = ("umask 077 && "
 436                "cat > '%s' << '!EOF.' && \n"
 437                "%s!EOF.\n"
 438                "cat > '%s' << '!EOF.' && \n"
 439                "%s!EOF.\n"
 440                "cat > '%s' << '!EOF.' && \n"
 441                "%s!EOF.\n"
 442                "chmod 0400 %s %s %s && "
 443                "%s start %s" %
 444                (constants.NODED_CERT_FILE, noded_cert,
 445                 constants.RAPI_CERT_FILE, rapi_cert,
 446                 constants.CONFD_HMAC_KEY, confd_hmac_key,
 447                 constants.NODED_CERT_FILE, constants.RAPI_CERT_FILE,
 448                 constants.CONFD_HMAC_KEY,
 449                 constants.DAEMON_UTIL, constants.NODED))
 450
 451   result = sshrunner.Run(node, 'root', mycommand, batch=False,
 452                          ask_key=ssh_key_check,
 453                          use_cluster_key=False,
 454                          strict_host_check=ssh_key_check)
 455   if result.failed:
 456     raise errors.OpExecError("Remote command on node %s, error: %s,"
 457                              " output: %s" %
 458                              (node, result.fail_reason, result.output))
 459
 460   _WaitForNodeDaemon(node)
 461
 462
 463 def MasterFailover(no_voting=False):
 464   """Failover the master node.
 465
 466   This checks that we are not already the master, and will cause the
 467   current master to cease being master, and the non-master to become
 468   new master.
 469
 470   @type no_voting: boolean
 471   @param no_voting: force the operation without remote nodes agreement
 472                       (dangerous)
 473
 474   """
 475   sstore = ssconf.SimpleStore()
 476
 477   old_master, new_master = ssconf.GetMasterAndMyself(sstore)
 478   node_list = sstore.GetNodeList()
 479   mc_list = sstore.GetMasterCandidates()
 480
 481   if old_master == new_master:
 482     raise errors.OpPrereqError("This commands must be run on the node"
 483                                " where you want the new master to be."
 484                                " %s is already the master" %
 485                                old_master, errors.ECODE_INVAL)
 486
 487   if new_master not in mc_list:
 488     mc_no_master = [name for name in mc_list if name != old_master]
 489     raise errors.OpPrereqError("This node is not among the nodes marked"
 490                                " as master candidates. Only these nodes"
 491                                " can become masters. Current list of"
 492                                " master candidates is:\n"
 493                                "%s" % ('\n'.join(mc_no_master)),
 494                                errors.ECODE_STATE)
 495
 496   if not no_voting:
 497     vote_list = GatherMasterVotes(node_list)
 498
 499     if vote_list:
 500       voted_master = vote_list[0][0]
 501       if voted_master is None:
 502         raise errors.OpPrereqError("Cluster is inconsistent, most nodes did"
 503                                    " not respond.", errors.ECODE_ENVIRON)
 504       elif voted_master != old_master:
 505         raise errors.OpPrereqError("I have a wrong configuration, I believe"
 506                                    " the master is %s but the other nodes"
 507                                    " voted %s. Please resync the configuration"
 508                                    " of this node." %
 509                                    (old_master, voted_master),
 510                                    errors.ECODE_STATE)
 511   # end checks
 512
 513   rcode = 0
 514
 515   logging.info("Setting master to %s, old master: %s", new_master, old_master)
 516
 517   result = rpc.RpcRunner.call_node_stop_master(old_master, True)
 518   msg = result.fail_msg
 519   if msg:
 520     logging.error("Could not disable the master role on the old master"
 521                  " %s, please disable manually: %s", old_master, msg)
 522
 523   # Here we have a phase where no master should be running
 524
 525   # instantiate a real config writer, as we now know we have the
 526   # configuration data
 527   cfg = config.ConfigWriter()
 528
 529   cluster_info = cfg.GetClusterInfo()
 530   cluster_info.master_node = new_master
 531   # this will also regenerate the ssconf files, since we updated the
 532   # cluster info
 533   cfg.Update(cluster_info, logging.error)
 534
 535   result = rpc.RpcRunner.call_node_start_master(new_master, True, no_voting)
 536   msg = result.fail_msg
 537   if msg:
 538     logging.error("Could not start the master role on the new master"
 539                   " %s, please check: %s", new_master, msg)
 540     rcode = 1
 541
 542   return rcode
 543
 544
 545 def GetMaster():
 546   """Returns the current master node.
 547
 548   This is a separate function in bootstrap since it's needed by
 549   gnt-cluster, and instead of importing directly ssconf, it's better
 550   to abstract it in bootstrap, where we do use ssconf in other
 551   functions too.
 552
 553   """
 554   sstore = ssconf.SimpleStore()
 555
 556   old_master, _ = ssconf.GetMasterAndMyself(sstore)
 557
 558   return old_master
 559
 560
 561 def GatherMasterVotes(node_list):
 562   """Check the agreement on who is the master.
 563
 564   This function will return a list of (node, number of votes), ordered
 565   by the number of votes. Errors will be denoted by the key 'None'.
 566
 567   Note that the sum of votes is the number of nodes this machine
 568   knows, whereas the number of entries in the list could be different
 569   (if some nodes vote for another master).
 570
 571   We remove ourselves from the list since we know that (bugs aside)
 572   since we use the same source for configuration information for both
 573   backend and boostrap, we'll always vote for ourselves.
 574
 575   @type node_list: list
 576   @param node_list: the list of nodes to query for master info; the current
 577       node will be removed if it is in the list
 578   @rtype: list
 579   @return: list of (node, votes)
 580
 581   """
 582   myself = utils.HostInfo().name
 583   try:
 584     node_list.remove(myself)
 585   except ValueError:
 586     pass
 587   if not node_list:
 588     # no nodes left (eventually after removing myself)
 589     return []
 590   results = rpc.RpcRunner.call_master_info(node_list)
 591   if not isinstance(results, dict):
 592     # this should not happen (unless internal error in rpc)
 593     logging.critical("Can't complete rpc call, aborting master startup")
 594     return [(None, len(node_list))]
 595   votes = {}
 596   for node in results:
 597     nres = results[node]
 598     data = nres.payload
 599     msg = nres.fail_msg
 600     fail = False
 601     if msg:
 602       logging.warning("Error contacting node %s: %s", node, msg)
 603       fail = True
 604     elif not isinstance(data, (tuple, list)) or len(data) < 3:
 605       logging.warning("Invalid data received from node %s: %s", node, data)
 606       fail = True
 607     if fail:
 608       if None not in votes:
 609         votes[None] = 0
 610       votes[None] += 1
 611       continue
 612     master_node = data[2]
 613     if master_node not in votes:
 614       votes[master_node] = 0
 615     votes[master_node] += 1
 616
 617   vote_list = [v for v in votes.items()]
 618   # sort first on number of votes then on name, since we want None
 619   # sorted later if we have the half of the nodes not responding, and
 620   # half voting all for the same master
 621   vote_list.sort(key=lambda x: (x[1], x[0]), reverse=True)
 622
 623   return vote_list