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

   1 #
   2 #
   3
   4 # Copyright (C) 2006, 2007, 2010, 2011, 2012 Google Inc.
   5 #
   6 # This program is free software; you can redistribute it and/or modify
   7 # it under the terms of the GNU General Public License as published by
   8 # the Free Software Foundation; either version 2 of the License, or
   9 # (at your option) any later version.
  10 #
  11 # This program is distributed in the hope that it will be useful, but
  12 # WITHOUT ANY WARRANTY; without even the implied warranty of
  13 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  14 # General Public License for more details.
  15 #
  16 # You should have received a copy of the GNU General Public License
  17 # along with this program; if not, write to the Free Software
  18 # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
  19 # 02110-1301, USA.
  20
  21
  22 """Master daemon program.
  23
  24 Some classes deviates from the standard style guide since the
  25 inheritance from parent classes requires it.
  26
  27 """
  28
  29 # pylint: disable=C0103
  30 # C0103: Invalid name ganeti-masterd
  31
  32 import grp
  33 import os
  34 import pwd
  35 import sys
  36 import socket
  37 import time
  38 import tempfile
  39 import logging
  40
  41 from optparse import OptionParser
  42
  43 from ganeti import config
  44 from ganeti import constants
  45 from ganeti import daemon
  46 from ganeti import mcpu
  47 from ganeti import opcodes
  48 from ganeti import jqueue
  49 from ganeti import locking
  50 from ganeti import luxi
  51 from ganeti import utils
  52 from ganeti import errors
  53 from ganeti import ssconf
  54 from ganeti import workerpool
  55 from ganeti import rpc
  56 from ganeti import bootstrap
  57 from ganeti import netutils
  58 from ganeti import objects
  59 from ganeti import query
  60 from ganeti import runtime
  61
  62
  63 CLIENT_REQUEST_WORKERS = 16
  64
  65 EXIT_NOTMASTER = constants.EXIT_NOTMASTER
  66 EXIT_NODESETUP_ERROR = constants.EXIT_NODESETUP_ERROR
  67
  68
  69 class ClientRequestWorker(workerpool.BaseWorker):
  70   # pylint: disable=W0221
  71   def RunTask(self, server, message, client):
  72     """Process the request.
  73
  74     """
  75     client_ops = ClientOps(server)
  76
  77     try:
  78       (method, args, version) = luxi.ParseRequest(message)
  79     except luxi.ProtocolError, err:
  80       logging.error("Protocol Error: %s", err)
  81       client.close_log()
  82       return
  83
  84     success = False
  85     try:
  86       # Verify client's version if there was one in the request
  87       if version is not None and version != constants.LUXI_VERSION:
  88         raise errors.LuxiError("LUXI version mismatch, server %s, request %s" %
  89                                (constants.LUXI_VERSION, version))
  90
  91       result = client_ops.handle_request(method, args)
  92       success = True
  93     except errors.GenericError, err:
  94       logging.exception("Unexpected exception")
  95       success = False
  96       result = errors.EncodeException(err)
  97     except:
  98       logging.exception("Unexpected exception")
  99       err = sys.exc_info()
 100       result = "Caught exception: %s" % str(err[1])
 101
 102     try:
 103       reply = luxi.FormatResponse(success, result)
 104       client.send_message(reply)
 105       # awake the main thread so that it can write out the data.
 106       server.awaker.signal()
 107     except: # pylint: disable=W0702
 108       logging.exception("Send error")
 109       client.close_log()
 110
 111
 112 class MasterClientHandler(daemon.AsyncTerminatedMessageStream):
 113   """Handler for master peers.
 114
 115   """
 116   _MAX_UNHANDLED = 1
 117
 118   def __init__(self, server, connected_socket, client_address, family):
 119     daemon.AsyncTerminatedMessageStream.__init__(self, connected_socket,
 120                                                  client_address,
 121                                                  constants.LUXI_EOM,
 122                                                  family, self._MAX_UNHANDLED)
 123     self.server = server
 124
 125   def handle_message(self, message, _):
 126     self.server.request_workers.AddTask((self.server, message, self))
 127
 128
 129 class _MasterShutdownCheck:
 130   """Logic for master daemon shutdown.
 131
 132   """
 133   #: How long to wait between checks
 134   _CHECK_INTERVAL = 5.0
 135
 136   #: How long to wait after all jobs are done (e.g. to give clients time to
 137   #: retrieve the job status)
 138   _SHUTDOWN_LINGER = 5.0
 139
 140   def __init__(self):
 141     """Initializes this class.
 142
 143     """
 144     self._had_active_jobs = None
 145     self._linger_timeout = None
 146
 147   def __call__(self, jq_prepare_result):
 148     """Determines if master daemon is ready for shutdown.
 149
 150     @param jq_prepare_result: Result of L{jqueue.JobQueue.PrepareShutdown}
 151     @rtype: None or number
 152     @return: None if master daemon is ready, timeout if the check must be
 153              repeated
 154
 155     """
 156     if jq_prepare_result:
 157       # Check again shortly
 158       logging.info("Job queue has been notified for shutdown but is still"
 159                    " busy; next check in %s seconds", self._CHECK_INTERVAL)
 160       self._had_active_jobs = True
 161       return self._CHECK_INTERVAL
 162
 163     if not self._had_active_jobs:
 164       # Can shut down as there were no active jobs on the first check
 165       return None
 166
 167     # No jobs are running anymore, but maybe some clients want to collect some
 168     # information. Give them a short amount of time.
 169     if self._linger_timeout is None:
 170       self._linger_timeout = utils.RunningTimeout(self._SHUTDOWN_LINGER, True)
 171
 172     remaining = self._linger_timeout.Remaining()
 173
 174     logging.info("Job queue no longer busy; shutting down master daemon"
 175                  " in %s seconds", remaining)
 176
 177     # TODO: Should the master daemon socket be closed at this point? Doing so
 178     # wouldn't affect existing connections.
 179
 180     if remaining < 0:
 181       return None
 182     else:
 183       return remaining
 184
 185
 186 class MasterServer(daemon.AsyncStreamServer):
 187   """Master Server.
 188
 189   This is the main asynchronous master server. It handles connections to the
 190   master socket.
 191
 192   """
 193   family = socket.AF_UNIX
 194
 195   def __init__(self, address, uid, gid):
 196     """MasterServer constructor
 197
 198     @param address: the unix socket address to bind the MasterServer to
 199     @param uid: The uid of the owner of the socket
 200     @param gid: The gid of the owner of the socket
 201
 202     """
 203     temp_name = tempfile.mktemp(dir=os.path.dirname(address))
 204     daemon.AsyncStreamServer.__init__(self, self.family, temp_name)
 205     os.chmod(temp_name, 0770)
 206     os.chown(temp_name, uid, gid)
 207     os.rename(temp_name, address)
 208
 209     self.awaker = daemon.AsyncAwaker()
 210
 211     # We'll only start threads once we've forked.
 212     self.context = None
 213     self.request_workers = None
 214
 215     self._shutdown_check = None
 216
 217   def handle_connection(self, connected_socket, client_address):
 218     # TODO: add connection count and limit the number of open connections to a
 219     # maximum number to avoid breaking for lack of file descriptors or memory.
 220     MasterClientHandler(self, connected_socket, client_address, self.family)
 221
 222   def setup_queue(self):
 223     self.context = GanetiContext()
 224     self.request_workers = workerpool.WorkerPool("ClientReq",
 225                                                  CLIENT_REQUEST_WORKERS,
 226                                                  ClientRequestWorker)
 227
 228   def WaitForShutdown(self):
 229     """Prepares server for shutdown.
 230
 231     """
 232     if self._shutdown_check is None:
 233       self._shutdown_check = _MasterShutdownCheck()
 234
 235     return self._shutdown_check(self.context.jobqueue.PrepareShutdown())
 236
 237   def server_cleanup(self):
 238     """Cleanup the server.
 239
 240     This involves shutting down the processor threads and the master
 241     socket.
 242
 243     """
 244     try:
 245       self.close()
 246     finally:
 247       if self.request_workers:
 248         self.request_workers.TerminateWorkers()
 249       if self.context:
 250         self.context.jobqueue.Shutdown()
 251
 252
 253 class ClientOps:
 254   """Class holding high-level client operations."""
 255   def __init__(self, server):
 256     self.server = server
 257
 258   def handle_request(self, method, args): # pylint: disable=R0911
 259     context = self.server.context
 260     queue = context.jobqueue
 261
 262     # TODO: Parameter validation
 263     if not isinstance(args, (tuple, list)):
 264       logging.info("Received invalid arguments of type '%s'", type(args))
 265       raise ValueError("Invalid arguments type '%s'" % type(args))
 266
 267     # TODO: Rewrite to not exit in each 'if/elif' branch
 268
 269     if method == luxi.REQ_SUBMIT_JOB:
 270       logging.info("Received new job")
 271       (job_def, ) = args
 272       ops = [opcodes.OpCode.LoadOpCode(state) for state in job_def]
 273       return queue.SubmitJob(ops)
 274
 275     elif method == luxi.REQ_SUBMIT_MANY_JOBS:
 276       logging.info("Received multiple jobs")
 277       (job_defs, ) = args
 278       jobs = []
 279       for ops in job_defs:
 280         jobs.append([opcodes.OpCode.LoadOpCode(state) for state in ops])
 281       return queue.SubmitManyJobs(jobs)
 282
 283     elif method == luxi.REQ_CANCEL_JOB:
 284       (job_id, ) = args
 285       logging.info("Received job cancel request for %s", job_id)
 286       return queue.CancelJob(job_id)
 287
 288     elif method == luxi.REQ_ARCHIVE_JOB:
 289       (job_id, ) = args
 290       logging.info("Received job archive request for %s", job_id)
 291       return queue.ArchiveJob(job_id)
 292
 293     elif method == luxi.REQ_AUTO_ARCHIVE_JOBS:
 294       (age, timeout) = args
 295       logging.info("Received job autoarchive request for age %s, timeout %s",
 296                    age, timeout)
 297       return queue.AutoArchiveJobs(age, timeout)
 298
 299     elif method == luxi.REQ_WAIT_FOR_JOB_CHANGE:
 300       (job_id, fields, prev_job_info, prev_log_serial, timeout) = args
 301       logging.info("Received job poll request for %s", job_id)
 302       return queue.WaitForJobChanges(job_id, fields, prev_job_info,
 303                                      prev_log_serial, timeout)
 304
 305     elif method == luxi.REQ_QUERY:
 306       (what, fields, qfilter) = args
 307
 308       if what in constants.QR_VIA_OP:
 309         result = self._Query(opcodes.OpQuery(what=what, fields=fields,
 310                                              qfilter=qfilter))
 311       elif what == constants.QR_LOCK:
 312         if qfilter is not None:
 313           raise errors.OpPrereqError("Lock queries can't be filtered",
 314                                      errors.ECODE_INVAL)
 315         return context.glm.QueryLocks(fields)
 316       elif what == constants.QR_JOB:
 317         return queue.QueryJobs(fields, qfilter)
 318       elif what in constants.QR_VIA_LUXI:
 319         raise NotImplementedError
 320       else:
 321         raise errors.OpPrereqError("Resource type '%s' unknown" % what,
 322                                    errors.ECODE_INVAL)
 323
 324       return result
 325
 326     elif method == luxi.REQ_QUERY_FIELDS:
 327       (what, fields) = args
 328       req = objects.QueryFieldsRequest(what=what, fields=fields)
 329
 330       try:
 331         fielddefs = query.ALL_FIELDS[req.what]
 332       except KeyError:
 333         raise errors.OpPrereqError("Resource type '%s' unknown" % req.what,
 334                                    errors.ECODE_INVAL)
 335
 336       return query.QueryFields(fielddefs, req.fields)
 337
 338     elif method == luxi.REQ_QUERY_JOBS:
 339       (job_ids, fields) = args
 340       if isinstance(job_ids, (tuple, list)) and job_ids:
 341         msg = utils.CommaJoin(job_ids)
 342       else:
 343         msg = str(job_ids)
 344       logging.info("Received job query request for %s", msg)
 345       return queue.OldStyleQueryJobs(job_ids, fields)
 346
 347     elif method == luxi.REQ_QUERY_INSTANCES:
 348       (names, fields, use_locking) = args
 349       logging.info("Received instance query request for %s", names)
 350       if use_locking:
 351         raise errors.OpPrereqError("Sync queries are not allowed",
 352                                    errors.ECODE_INVAL)
 353       op = opcodes.OpInstanceQuery(names=names, output_fields=fields,
 354                                    use_locking=use_locking)
 355       return self._Query(op)
 356
 357     elif method == luxi.REQ_QUERY_NODES:
 358       (names, fields, use_locking) = args
 359       logging.info("Received node query request for %s", names)
 360       if use_locking:
 361         raise errors.OpPrereqError("Sync queries are not allowed",
 362                                    errors.ECODE_INVAL)
 363       op = opcodes.OpNodeQuery(names=names, output_fields=fields,
 364                                use_locking=use_locking)
 365       return self._Query(op)
 366
 367     elif method == luxi.REQ_QUERY_GROUPS:
 368       (names, fields, use_locking) = args
 369       logging.info("Received group query request for %s", names)
 370       if use_locking:
 371         raise errors.OpPrereqError("Sync queries are not allowed",
 372                                    errors.ECODE_INVAL)
 373       op = opcodes.OpGroupQuery(names=names, output_fields=fields)
 374       return self._Query(op)
 375
 376     elif method == luxi.REQ_QUERY_EXPORTS:
 377       (nodes, use_locking) = args
 378       if use_locking:
 379         raise errors.OpPrereqError("Sync queries are not allowed",
 380                                    errors.ECODE_INVAL)
 381       logging.info("Received exports query request")
 382       op = opcodes.OpBackupQuery(nodes=nodes, use_locking=use_locking)
 383       return self._Query(op)
 384
 385     elif method == luxi.REQ_QUERY_CONFIG_VALUES:
 386       (fields, ) = args
 387       logging.info("Received config values query request for %s", fields)
 388       op = opcodes.OpClusterConfigQuery(output_fields=fields)
 389       return self._Query(op)
 390
 391     elif method == luxi.REQ_QUERY_CLUSTER_INFO:
 392       logging.info("Received cluster info query request")
 393       op = opcodes.OpClusterQuery()
 394       return self._Query(op)
 395
 396     elif method == luxi.REQ_QUERY_TAGS:
 397       (kind, name) = args
 398       logging.info("Received tags query request")
 399       op = opcodes.OpTagsGet(kind=kind, name=name, use_locking=False)
 400       return self._Query(op)
 401
 402     elif method == luxi.REQ_SET_DRAIN_FLAG:
 403       (drain_flag, ) = args
 404       logging.info("Received queue drain flag change request to %s",
 405                    drain_flag)
 406       return queue.SetDrainFlag(drain_flag)
 407
 408     elif method == luxi.REQ_SET_WATCHER_PAUSE:
 409       (until, ) = args
 410
 411       if until is None:
 412         logging.info("Received request to no longer pause the watcher")
 413       else:
 414         if not isinstance(until, (int, float)):
 415           raise TypeError("Duration must be an integer or float")
 416
 417         if until < time.time():
 418           raise errors.GenericError("Unable to set pause end time in the past")
 419
 420         logging.info("Received request to pause the watcher until %s", until)
 421
 422       return _SetWatcherPause(until)
 423
 424     else:
 425       logging.info("Received invalid request '%s'", method)
 426       raise ValueError("Invalid operation '%s'" % method)
 427
 428   def _Query(self, op):
 429     """Runs the specified opcode and returns the result.
 430
 431     """
 432     # Queries don't have a job id
 433     proc = mcpu.Processor(self.server.context, None, enable_locks=False)
 434
 435     # TODO: Executing an opcode using locks will acquire them in blocking mode.
 436     # Consider using a timeout for retries.
 437     return proc.ExecOpCode(op, None)
 438
 439
 440 class GanetiContext(object):
 441   """Context common to all ganeti threads.
 442
 443   This class creates and holds common objects shared by all threads.
 444
 445   """
 446   # pylint: disable=W0212
 447   # we do want to ensure a singleton here
 448   _instance = None
 449
 450   def __init__(self):
 451     """Constructs a new GanetiContext object.
 452
 453     There should be only a GanetiContext object at any time, so this
 454     function raises an error if this is not the case.
 455
 456     """
 457     assert self.__class__._instance is None, "double GanetiContext instance"
 458
 459     # Create global configuration object
 460     self.cfg = config.ConfigWriter()
 461
 462     # Locking manager
 463     self.glm = locking.GanetiLockManager(
 464                 self.cfg.GetNodeList(),
 465                 self.cfg.GetNodeGroupList(),
 466                 self.cfg.GetInstanceList())
 467
 468     self.cfg.SetContext(self)
 469
 470     # RPC runner
 471     self.rpc = rpc.RpcRunner(self.cfg, self.glm.AddToLockMonitor)
 472
 473     # Job queue
 474     self.jobqueue = jqueue.JobQueue(self)
 475
 476     # setting this also locks the class against attribute modifications
 477     self.__class__._instance = self
 478
 479   def __setattr__(self, name, value):
 480     """Setting GanetiContext attributes is forbidden after initialization.
 481
 482     """
 483     assert self.__class__._instance is None, "Attempt to modify Ganeti Context"
 484     object.__setattr__(self, name, value)
 485
 486   def AddNode(self, node, ec_id):
 487     """Adds a node to the configuration and lock manager.
 488
 489     """
 490     # Add it to the configuration
 491     self.cfg.AddNode(node, ec_id)
 492
 493     # If preseeding fails it'll not be added
 494     self.jobqueue.AddNode(node)
 495
 496     # Add the new node to the Ganeti Lock Manager
 497     self.glm.add(locking.LEVEL_NODE, node.name)
 498     self.glm.add(locking.LEVEL_NODE_RES, node.name)
 499
 500   def ReaddNode(self, node):
 501     """Updates a node that's already in the configuration
 502
 503     """
 504     # Synchronize the queue again
 505     self.jobqueue.AddNode(node)
 506
 507   def RemoveNode(self, name):
 508     """Removes a node from the configuration and lock manager.
 509
 510     """
 511     # Remove node from configuration
 512     self.cfg.RemoveNode(name)
 513
 514     # Notify job queue
 515     self.jobqueue.RemoveNode(name)
 516
 517     # Remove the node from the Ganeti Lock Manager
 518     self.glm.remove(locking.LEVEL_NODE, name)
 519     self.glm.remove(locking.LEVEL_NODE_RES, name)
 520
 521
 522 def _SetWatcherPause(until):
 523   """Creates or removes the watcher pause file.
 524
 525   @type until: None or int
 526   @param until: Unix timestamp saying until when the watcher shouldn't run
 527
 528   """
 529   if until is None:
 530     utils.RemoveFile(constants.WATCHER_PAUSEFILE)
 531   else:
 532     utils.WriteFile(constants.WATCHER_PAUSEFILE,
 533                     data="%d\n" % (until, ))
 534
 535   return until
 536
 537
 538 @rpc.RunWithRPC
 539 def CheckAgreement():
 540   """Check the agreement on who is the master.
 541
 542   The function uses a very simple algorithm: we must get more positive
 543   than negative answers. Since in most of the cases we are the master,
 544   we'll use our own config file for getting the node list. In the
 545   future we could collect the current node list from our (possibly
 546   obsolete) known nodes.
 547
 548   In order to account for cold-start of all nodes, we retry for up to
 549   a minute until we get a real answer as the top-voted one. If the
 550   nodes are more out-of-sync, for now manual startup of the master
 551   should be attempted.
 552
 553   Note that for a even number of nodes cluster, we need at least half
 554   of the nodes (beside ourselves) to vote for us. This creates a
 555   problem on two-node clusters, since in this case we require the
 556   other node to be up too to confirm our status.
 557
 558   """
 559   myself = netutils.Hostname.GetSysName()
 560   #temp instantiation of a config writer, used only to get the node list
 561   cfg = config.ConfigWriter()
 562   node_list = cfg.GetNodeList()
 563   del cfg
 564   retries = 6
 565   while retries > 0:
 566     votes = bootstrap.GatherMasterVotes(node_list)
 567     if not votes:
 568       # empty node list, this is a one node cluster
 569       return True
 570     if votes[0][0] is None:
 571       retries -= 1
 572       time.sleep(10)
 573       continue
 574     break
 575   if retries == 0:
 576     logging.critical("Cluster inconsistent, most of the nodes didn't answer"
 577                      " after multiple retries. Aborting startup")
 578     logging.critical("Use the --no-voting option if you understand what"
 579                      " effects it has on the cluster state")
 580     return False
 581   # here a real node is at the top of the list
 582   all_votes = sum(item[1] for item in votes)
 583   top_node, top_votes = votes[0]
 584
 585   result = False
 586   if top_node != myself:
 587     logging.critical("It seems we are not the master (top-voted node"
 588                      " is %s with %d out of %d votes)", top_node, top_votes,
 589                      all_votes)
 590   elif top_votes < all_votes - top_votes:
 591     logging.critical("It seems we are not the master (%d votes for,"
 592                      " %d votes against)", top_votes, all_votes - top_votes)
 593   else:
 594     result = True
 595
 596   return result
 597
 598
 599 @rpc.RunWithRPC
 600 def ActivateMasterIP():
 601   # activate ip
 602   cfg = config.ConfigWriter()
 603   master_params = cfg.GetMasterNetworkParameters()
 604   ems = cfg.GetUseExternalMipScript()
 605   runner = rpc.BootstrapRunner()
 606   result = runner.call_node_activate_master_ip(master_params.name,
 607                                                master_params, ems)
 608
 609   msg = result.fail_msg
 610   if msg:
 611     logging.error("Can't activate master IP address: %s", msg)
 612
 613
 614 def CheckMasterd(options, args):
 615   """Initial checks whether to run or exit with a failure.
 616
 617   """
 618   if args: # masterd doesn't take any arguments
 619     print >> sys.stderr, ("Usage: %s [-f] [-d]" % sys.argv[0])
 620     sys.exit(constants.EXIT_FAILURE)
 621
 622   ssconf.CheckMaster(options.debug)
 623
 624   try:
 625     options.uid = pwd.getpwnam(constants.MASTERD_USER).pw_uid
 626     options.gid = grp.getgrnam(constants.DAEMONS_GROUP).gr_gid
 627   except KeyError:
 628     print >> sys.stderr, ("User or group not existing on system: %s:%s" %
 629                           (constants.MASTERD_USER, constants.DAEMONS_GROUP))
 630     sys.exit(constants.EXIT_FAILURE)
 631
 632   # Determine static runtime architecture information
 633   runtime.InitArchInfo()
 634
 635   # Check the configuration is sane before anything else
 636   try:
 637     config.ConfigWriter()
 638   except errors.ConfigVersionMismatch, err:
 639     v1 = "%s.%s.%s" % constants.SplitVersion(err.args[0])
 640     v2 = "%s.%s.%s" % constants.SplitVersion(err.args[1])
 641     print >> sys.stderr,  \
 642         ("Configuration version mismatch. The current Ganeti software"
 643          " expects version %s, but the on-disk configuration file has"
 644          " version %s. This is likely the result of upgrading the"
 645          " software without running the upgrade procedure. Please contact"
 646          " your cluster administrator or complete the upgrade using the"
 647          " cfgupgrade utility, after reading the upgrade notes." %
 648          (v1, v2))
 649     sys.exit(constants.EXIT_FAILURE)
 650   except errors.ConfigurationError, err:
 651     print >> sys.stderr, \
 652         ("Configuration error while opening the configuration file: %s\n"
 653          "This might be caused by an incomplete software upgrade or"
 654          " by a corrupted configuration file. Until the problem is fixed"
 655          " the master daemon cannot start." % str(err))
 656     sys.exit(constants.EXIT_FAILURE)
 657
 658   # If CheckMaster didn't fail we believe we are the master, but we have to
 659   # confirm with the other nodes.
 660   if options.no_voting:
 661     if not options.yes_do_it:
 662       sys.stdout.write("The 'no voting' option has been selected.\n")
 663       sys.stdout.write("This is dangerous, please confirm by"
 664                        " typing uppercase 'yes': ")
 665       sys.stdout.flush()
 666
 667       confirmation = sys.stdin.readline().strip()
 668       if confirmation != "YES":
 669         print >> sys.stderr, "Aborting."
 670         sys.exit(constants.EXIT_FAILURE)
 671
 672   else:
 673     # CheckAgreement uses RPC and threads, hence it needs to be run in
 674     # a separate process before we call utils.Daemonize in the current
 675     # process.
 676     if not utils.RunInSeparateProcess(CheckAgreement):
 677       sys.exit(constants.EXIT_FAILURE)
 678
 679   # ActivateMasterIP also uses RPC/threads, so we run it again via a
 680   # separate process.
 681
 682   # TODO: decide whether failure to activate the master IP is a fatal error
 683   utils.RunInSeparateProcess(ActivateMasterIP)
 684
 685
 686 def PrepMasterd(options, _):
 687   """Prep master daemon function, executed with the PID file held.
 688
 689   """
 690   # This is safe to do as the pid file guarantees against
 691   # concurrent execution.
 692   utils.RemoveFile(constants.MASTER_SOCKET)
 693
 694   mainloop = daemon.Mainloop()
 695   master = MasterServer(constants.MASTER_SOCKET, options.uid, options.gid)
 696   return (mainloop, master)
 697
 698
 699 def ExecMasterd(options, args, prep_data): # pylint: disable=W0613
 700   """Main master daemon function, executed with the PID file held.
 701
 702   """
 703   (mainloop, master) = prep_data
 704   try:
 705     rpc.Init()
 706     try:
 707       master.setup_queue()
 708       try:
 709         mainloop.Run(shutdown_wait_fn=master.WaitForShutdown)
 710       finally:
 711         master.server_cleanup()
 712     finally:
 713       rpc.Shutdown()
 714   finally:
 715     utils.RemoveFile(constants.MASTER_SOCKET)
 716
 717   logging.info("Clean master daemon shutdown")
 718
 719
 720 def Main():
 721   """Main function"""
 722   parser = OptionParser(description="Ganeti master daemon",
 723                         usage="%prog [-f] [-d]",
 724                         version="%%prog (ganeti) %s" %
 725                         constants.RELEASE_VERSION)
 726   parser.add_option("--no-voting", dest="no_voting",
 727                     help="Do not check that the nodes agree on this node"
 728                     " being the master and start the daemon unconditionally",
 729                     default=False, action="store_true")
 730   parser.add_option("--yes-do-it", dest="yes_do_it",
 731                     help="Override interactive check for --no-voting",
 732                     default=False, action="store_true")
 733   daemon.GenericMain(constants.MASTERD, parser, CheckMasterd, PrepMasterd,
 734                      ExecMasterd, multithreaded=True)