import sys
import SocketServer
-import threading
import time
import collections
import Queue
from ganeti import errors
from ganeti import ssconf
from ganeti import logger
+from ganeti import workerpool
+from ganeti import rpc
+CLIENT_REQUEST_WORKERS = 16
+
EXIT_NOTMASTER = constants.EXIT_NOTMASTER
EXIT_NODESETUP_ERROR = constants.EXIT_NODESETUP_ERROR
+class ClientRequestWorker(workerpool.BaseWorker):
+ def RunTask(self, server, request, client_address):
+ """Process the request.
+
+ This is copied from the code in ThreadingMixIn.
+
+ """
+ try:
+ server.finish_request(request, client_address)
+ server.close_request(request)
+ except:
+ server.handle_error(request, client_address)
+ server.close_request(request)
+
+
class IOServer(SocketServer.UnixStreamServer):
"""IO thread class.
"""
SocketServer.UnixStreamServer.__init__(self, address, rqhandler)
- self.do_quit = False
self.context = context
# We'll only start threads once we've forked.
self.jobqueue = None
-
- signal.signal(signal.SIGINT, self.handle_quit_signals)
- signal.signal(signal.SIGTERM, self.handle_quit_signals)
+ self.request_workers = None
def setup_queue(self):
self.jobqueue = jqueue.JobQueue(self.context)
-
- def process_request_thread(self, request, client_address):
- """Process the request.
-
- This is copied from the code in ThreadingMixIn.
-
- """
- try:
- self.finish_request(request, client_address)
- self.close_request(request)
- except:
- self.handle_error(request, client_address)
- self.close_request(request)
+ self.request_workers = workerpool.WorkerPool(CLIENT_REQUEST_WORKERS,
+ ClientRequestWorker)
def process_request(self, request, client_address):
- """Start a new thread to process the request.
-
- This is copied from the coode in ThreadingMixIn.
+ """Add task to workerpool to process request.
"""
- t = threading.Thread(target=self.process_request_thread,
- args=(request, client_address))
- t.start()
-
- def handle_quit_signals(self, signum, frame):
- print "received %s in %s" % (signum, frame)
- self.do_quit = True
+ self.request_workers.AddTask(self, request, client_address)
def serve_forever(self):
"""Handle one request at a time until told to quit."""
- while not self.do_quit:
- self.handle_request()
- print "served request, quit=%s" % (self.do_quit)
+ sighandler = utils.SignalHandler([signal.SIGINT, signal.SIGTERM])
+ try:
+ while not sighandler.called:
+ self.handle_request()
+ finally:
+ sighandler.Reset()
def server_cleanup(self):
"""Cleanup the server.
self.server_close()
utils.RemoveFile(constants.MASTER_SOCKET)
finally:
+ if self.request_workers:
+ self.request_workers.TerminateWorkers()
if self.jobqueue:
self.jobqueue.Shutdown()
if method == luxi.REQ_SUBMIT_JOB:
ops = [opcodes.OpCode.LoadOpCode(state) for state in args]
- return queue.SubmitJob(ops)
+ # we need to compute the node list here, since from now on all
+ # operations require locks on the queue or the storage, and we
+ # shouldn't get another lock
+ node_list = self.server.context.cfg.GetNodeList()
+ return queue.SubmitJob(ops, node_list)
elif method == luxi.REQ_CANCEL_JOB:
- (job_id, ) = args
+ job_id = args
return queue.CancelJob(job_id)
elif method == luxi.REQ_ARCHIVE_JOB:
- (job_id, ) = args
+ job_id = args
return queue.ArchiveJob(job_id)
elif method == luxi.REQ_QUERY_JOBS:
raise ValueError("Invalid operation")
-def JobRunner(proc, job, context):
- """Job executor.
-
- This functions processes a single job in the context of given
- processor instance.
-
- Args:
- proc: Ganeti Processor to run the job on
- job: The job to run (unserialized format)
- context: Ganeti shared context
-
- """
- job.SetStatus(opcodes.Job.STATUS_RUNNING)
- fail = False
- for idx, op in enumerate(job.data.op_list):
- job.data.op_status[idx] = opcodes.Job.STATUS_RUNNING
- try:
- job.data.op_result[idx] = proc.ExecOpCode(op)
- job.data.op_status[idx] = opcodes.Job.STATUS_SUCCESS
- except (errors.OpPrereqError, errors.OpExecError), err:
- fail = True
- job.data.op_result[idx] = str(err)
- job.data.op_status[idx] = opcodes.Job.STATUS_FAIL
- if fail:
- job.SetStatus(opcodes.Job.STATUS_FAIL)
- else:
- job.SetStatus(opcodes.Job.STATUS_SUCCESS)
-
-
-def PoolWorker(worker_id, incoming_queue, context):
- """A worker thread function.
-
- This is the actual processor of a single thread of Job execution.
-
- Args:
- worker_id: the unique id for this worker
- incoming_queue: a queue to get jobs from
- context: the common server context, containing all shared data and
- synchronization structures.
-
- """
- while True:
- logging.debug("worker %s sleeping", worker_id)
- item = incoming_queue.get(True)
- if item is None:
- break
- logging.debug("worker %s processing job %s", worker_id, item.data.job_id)
- proc = mcpu.Processor(context, feedback=lambda x: None)
- try:
- JobRunner(proc, item, context)
- except errors.GenericError, err:
- msg = "ganeti exception"
- logging.error(msg, exc_info=err)
- item.SetStatus(opcodes.Job.STATUS_FAIL, result=[msg])
- except Exception, err:
- msg = "unhandled exception"
- logging.error(msg, exc_info=err)
- item.SetStatus(opcodes.Job.STATUS_FAIL, result=[msg])
- except:
- msg = "unhandled unknown exception"
- logging.error(msg, exc_info=True)
- item.SetStatus(opcodes.Job.STATUS_FAIL, result=[msg])
- logging.debug("worker %s finish job %s", worker_id, item.data.job_id)
- logging.debug("worker %s exiting", worker_id)
-
-
class GanetiContext(object):
"""Context common to all ganeti threads.
object.__setattr__(self, name, value)
-def CheckMaster(debug):
- """Checks the node setup.
-
- If this is the master, the function will return. Otherwise it will
- exit with an exit code based on the node status.
-
- """
- try:
- ss = ssconf.SimpleStore()
- master_name = ss.GetMasterNode()
- except errors.ConfigurationError, err:
- print "Cluster configuration incomplete: '%s'" % str(err)
- sys.exit(EXIT_NODESETUP_ERROR)
-
- try:
- myself = utils.HostInfo()
- except errors.ResolverError, err:
- sys.stderr.write("Cannot resolve my own name (%s)\n" % err.args[0])
- sys.exit(EXIT_NODESETUP_ERROR)
-
- if myself.name != master_name:
- if debug:
- sys.stderr.write("Not master, exiting.\n")
- sys.exit(EXIT_NOTMASTER)
-
-
def ParseOptions():
"""Parse the command line options.
utils.debug = options.debug
utils.no_fork = True
- CheckMaster(options.debug)
+ ssconf.CheckMaster(options.debug)
master = IOServer(constants.MASTER_SOCKET, ClientRqHandler, GanetiContext())
utils.Daemonize(logfile=constants.LOG_MASTERDAEMON,
noclose_fds=[master.fileno()])
- logger.SetupDaemon(constants.LOG_MASTERDAEMON, debug=options.debug,
- stderr_logging=not options.fork)
+ utils.WritePidFile(constants.MASTERD_PID)
+
+ logger.SetupLogging(constants.LOG_MASTERDAEMON, debug=options.debug,
+ stderr_logging=not options.fork)
logging.info("ganeti master daemon startup")
+ # activate ip
+ master_node = ssconf.SimpleStore().GetMasterNode()
+ if not rpc.call_node_start_master(master_node, False):
+ logging.error("Can't activate master IP address")
+
master.setup_queue()
try:
master.serve_forever()
finally:
master.server_cleanup()
+ utils.RemovePidFile(constants.MASTERD_PID)
if __name__ == "__main__":
projects / ganeti-local / blobdiff