X-Git-Url: https://code.grnet.gr/git/ganeti-local/blobdiff_plain/04ccf5e9f58a884eb2c949b2aabfcea33ad806a5..4478301b4396b9083217ccdb14e9a6b2825dc487:/lib/daemon.py diff --git a/lib/daemon.py b/lib/daemon.py index 1c63a1b..c3c4761 100644 --- a/lib/daemon.py +++ b/lib/daemon.py @@ -1,7 +1,7 @@ # # -# Copyright (C) 2006, 2007, 2008 Google Inc. +# Copyright (C) 2006, 2007, 2008, 2010 Google Inc. # # This program is free software; you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by @@ -22,237 +22,459 @@ """Module with helper classes and functions for daemons""" -import select +import asyncore +import asynchat +import collections +import os import signal -import errno -import time import logging +import sched +import time +import socket +import select +import sys from ganeti import utils from ganeti import constants +from ganeti import errors +from ganeti import netutils +from ganeti import ssconf +from ganeti import runtime -class Timer(object): - def __init__(self, owner, timer_id, start, interval, repeat): - self.owner = owner - self.timer_id = timer_id - self.start = start - self.interval = interval - self.repeat = repeat +class SchedulerBreakout(Exception): + """Exception used to get out of the scheduler loop + """ -class Mainloop(object): - """Generic mainloop for daemons + +def AsyncoreDelayFunction(timeout): + """Asyncore-compatible scheduler delay function. + + This is a delay function for sched that, rather than actually sleeping, + executes asyncore events happening in the meantime. + + After an event has occurred, rather than returning, it raises a + SchedulerBreakout exception, which will force the current scheduler.run() + invocation to terminate, so that we can also check for signals. The main loop + will then call the scheduler run again, which will allow it to actually + process any due events. + + This is needed because scheduler.run() doesn't support a count=..., as + asyncore loop, and the scheduler module documents throwing exceptions from + inside the delay function as an allowed usage model. """ - def __init__(self): - """Constructs a new Mainloop instance. + asyncore.loop(timeout=timeout, count=1, use_poll=True) + raise SchedulerBreakout() + + +class AsyncoreScheduler(sched.scheduler): + """Event scheduler integrated with asyncore + + """ + def __init__(self, timefunc): + sched.scheduler.__init__(self, timefunc, AsyncoreDelayFunction) + + +class GanetiBaseAsyncoreDispatcher(asyncore.dispatcher): + """Base Ganeti Asyncore Dispacher + + """ + # this method is overriding an asyncore.dispatcher method + def handle_error(self): + """Log an error in handling any request, and proceed. """ - self._io_wait = {} - self._io_wait_add = [] - self._io_wait_remove = [] - self._signal_wait = [] - self._timer_id_last = 0 - self._timer = {} - self._timer_add = [] - self._timer_remove = [] + logging.exception("Error while handling asyncore request") - def Run(self, handle_sigchld=True, handle_sigterm=True, stop_on_empty=False): - """Runs the mainloop. + # this method is overriding an asyncore.dispatcher method + def writable(self): + """Most of the time we don't want to check for writability. + + """ + return False + + +class AsyncStreamServer(GanetiBaseAsyncoreDispatcher): + """A stream server to use with asyncore. + + Each request is accepted, and then dispatched to a separate asyncore + dispatcher to handle. + + """ + + _REQUEST_QUEUE_SIZE = 5 - @type handle_sigchld: bool - @param handle_sigchld: Whether to install handler for SIGCHLD - @type handle_sigterm: bool - @param handle_sigterm: Whether to install handler for SIGTERM - @type stop_on_empty: bool - @param stop_on_empty: Whether to stop mainloop once all I/O waiters - unregistered + def __init__(self, family, address): + """Constructor for AsyncUnixStreamSocket + + @type family: integer + @param family: socket family (one of socket.AF_*) + @type address: address family dependent + @param address: address to bind the socket to + + """ + GanetiBaseAsyncoreDispatcher.__init__(self) + self.family = family + self.create_socket(self.family, socket.SOCK_STREAM) + self.set_reuse_addr() + self.bind(address) + self.listen(self._REQUEST_QUEUE_SIZE) + + # this method is overriding an asyncore.dispatcher method + def handle_accept(self): + """Accept a new client connection. + + Creates a new instance of the handler class, which will use asyncore to + serve the client. + + """ + accept_result = utils.IgnoreSignals(self.accept) + if accept_result is not None: + connected_socket, client_address = accept_result + if self.family == socket.AF_UNIX: + # override the client address, as for unix sockets nothing meaningful + # is passed in from accept anyway + client_address = netutils.GetSocketCredentials(connected_socket) + logging.info("Accepted connection from %s", + netutils.FormatAddress(client_address, family=self.family)) + self.handle_connection(connected_socket, client_address) + + def handle_connection(self, connected_socket, client_address): + """Handle an already accepted connection. """ - poller = select.poll() + raise NotImplementedError + - # Setup signal handlers - if handle_sigchld: - sigchld_handler = utils.SignalHandler([signal.SIGCHLD]) +class AsyncTerminatedMessageStream(asynchat.async_chat): + """A terminator separated message stream asyncore module. + + Handles a stream connection receiving messages terminated by a defined + separator. For each complete message handle_message is called. + + """ + def __init__(self, connected_socket, peer_address, terminator, family, + unhandled_limit): + """AsyncTerminatedMessageStream constructor. + + @type connected_socket: socket.socket + @param connected_socket: connected stream socket to receive messages from + @param peer_address: family-specific peer address + @type terminator: string + @param terminator: terminator separating messages in the stream + @type family: integer + @param family: socket family + @type unhandled_limit: integer or None + @param unhandled_limit: maximum unanswered messages + + """ + # python 2.4/2.5 uses conn=... while 2.6 has sock=... we have to cheat by + # using a positional argument rather than a keyword one. + asynchat.async_chat.__init__(self, connected_socket) + self.connected_socket = connected_socket + # on python 2.4 there is no "family" attribute for the socket class + # FIXME: when we move to python 2.5 or above remove the family parameter + #self.family = self.connected_socket.family + self.family = family + self.peer_address = peer_address + self.terminator = terminator + self.unhandled_limit = unhandled_limit + self.set_terminator(terminator) + self.ibuffer = [] + self.receive_count = 0 + self.send_count = 0 + self.oqueue = collections.deque() + self.iqueue = collections.deque() + + # this method is overriding an asynchat.async_chat method + def collect_incoming_data(self, data): + self.ibuffer.append(data) + + def _can_handle_message(self): + return (self.unhandled_limit is None or + (self.receive_count < self.send_count + self.unhandled_limit) and + not self.iqueue) + + # this method is overriding an asynchat.async_chat method + def found_terminator(self): + message = "".join(self.ibuffer) + self.ibuffer = [] + message_id = self.receive_count + # We need to increase the receive_count after checking if the message can + # be handled, but before calling handle_message + can_handle = self._can_handle_message() + self.receive_count += 1 + if can_handle: + self.handle_message(message, message_id) else: - sigchld_handler = None - try: - if handle_sigterm: - sigterm_handler = utils.SignalHandler([signal.SIGTERM]) + self.iqueue.append((message, message_id)) + + def handle_message(self, message, message_id): + """Handle a terminated message. + + @type message: string + @param message: message to handle + @type message_id: integer + @param message_id: stream's message sequence number + + """ + pass + # TODO: move this method to raise NotImplementedError + # raise NotImplementedError + + def send_message(self, message): + """Send a message to the remote peer. This function is thread-safe. + + @type message: string + @param message: message to send, without the terminator + + @warning: If calling this function from a thread different than the one + performing the main asyncore loop, remember that you have to wake that one + up. + + """ + # If we just append the message we received to the output queue, this + # function can be safely called by multiple threads at the same time, and + # we don't need locking, since deques are thread safe. handle_write in the + # asyncore thread will handle the next input message if there are any + # enqueued. + self.oqueue.append(message) + + # this method is overriding an asyncore.dispatcher method + def readable(self): + # read from the socket if we can handle the next requests + return self._can_handle_message() and asynchat.async_chat.readable(self) + + # this method is overriding an asyncore.dispatcher method + def writable(self): + # the output queue may become full just after we called writable. This only + # works if we know we'll have something else waking us up from the select, + # in such case, anyway. + return asynchat.async_chat.writable(self) or self.oqueue + + # this method is overriding an asyncore.dispatcher method + def handle_write(self): + if self.oqueue: + # if we have data in the output queue, then send_message was called. + # this means we can process one more message from the input queue, if + # there are any. + data = self.oqueue.popleft() + self.push(data + self.terminator) + self.send_count += 1 + if self.iqueue: + self.handle_message(*self.iqueue.popleft()) + self.initiate_send() + + def close_log(self): + logging.info("Closing connection from %s", + netutils.FormatAddress(self.peer_address, family=self.family)) + self.close() + + # this method is overriding an asyncore.dispatcher method + def handle_expt(self): + self.close_log() + + # this method is overriding an asyncore.dispatcher method + def handle_error(self): + """Log an error in handling any request, and proceed. + + """ + logging.exception("Error while handling asyncore request") + self.close_log() + + +class AsyncUDPSocket(GanetiBaseAsyncoreDispatcher): + """An improved asyncore udp socket. + + """ + def __init__(self, family): + """Constructor for AsyncUDPSocket + + """ + GanetiBaseAsyncoreDispatcher.__init__(self) + self._out_queue = [] + self._family = family + self.create_socket(family, socket.SOCK_DGRAM) + + # this method is overriding an asyncore.dispatcher method + def handle_connect(self): + # Python thinks that the first udp message from a source qualifies as a + # "connect" and further ones are part of the same connection. We beg to + # differ and treat all messages equally. + pass + + # this method is overriding an asyncore.dispatcher method + def handle_read(self): + recv_result = utils.IgnoreSignals(self.recvfrom, + constants.MAX_UDP_DATA_SIZE) + if recv_result is not None: + payload, address = recv_result + if self._family == socket.AF_INET6: + # we ignore 'flow info' and 'scope id' as we don't need them + ip, port, _, _ = address else: - sigterm_handler = None - - try: - running = True - timeout = None - timeout_needs_update = True - - # Start actual main loop - while running: - # Entries could be added again afterwards, hence removing first - if self._io_wait_remove: - for fd in self._io_wait_remove: - try: - poller.unregister(fd) - except KeyError: - pass - try: - del self._io_wait[fd] - except KeyError: - pass - self._io_wait_remove = [] - - # Add new entries - if self._io_wait_add: - for (owner, fd, conditions) in self._io_wait_add: - self._io_wait[fd] = owner - poller.register(fd, conditions) - self._io_wait_add = [] - - # Add new timers - if self._timer_add: - timeout_needs_update = True - for timer in self._timer_add: - self._timer[timer.timer_id] = timer - del self._timer_add[:] - - # Remove timers - if self._timer_remove: - timeout_needs_update = True - for timer_id in self._timer_remove: - try: - del self._timer[timer_id] - except KeyError: - pass - del self._timer_remove[:] - - # Stop if nothing is listening anymore - if stop_on_empty and not (self._io_wait or self._timer): - break - - # Calculate timeout again if required - if timeout_needs_update: - timeout = self._CalcTimeout(time.time()) - timeout_needs_update = False - - # Wait for I/O events - try: - io_events = poller.poll(timeout) - except select.error, err: - # EINTR can happen when signals are sent - if err.args and err.args[0] in (errno.EINTR,): - io_events = None - else: - raise - - after_poll = time.time() - - if io_events: - # Check for I/O events - for (evfd, evcond) in io_events: - owner = self._io_wait.get(evfd, None) - if owner: - owner.OnIO(evfd, evcond) - - if self._timer: - self._CheckTimers(after_poll) - - # Check whether signal was raised - if sigchld_handler and sigchld_handler.called: - self._CallSignalWaiters(signal.SIGCHLD) - sigchld_handler.Clear() - - if sigterm_handler and sigterm_handler.called: - self._CallSignalWaiters(signal.SIGTERM) - running = False - sigterm_handler.Clear() - finally: - # Restore signal handlers - if sigterm_handler: - sigterm_handler.Reset() - finally: - if sigchld_handler: - sigchld_handler.Reset() - - def _CalcTimeout(self, now): - if not self._timer: - return None - - timeout = None - - # TODO: Repeating timers - - min_timeout = 0.001 - - for timer in self._timer.itervalues(): - time_left = (timer.start + timer.interval) - now - if timeout is None or time_left < timeout: - timeout = time_left - if timeout < 0: - timeout = 0 - break - elif timeout < min_timeout: - timeout = min_timeout - break - - return timeout * 1000.0 - - def _CheckTimers(self, now): - # TODO: Repeating timers - for timer in self._timer.itervalues(): - if now < (timer.start + timer.interval): - continue - - timer.owner.OnTimer(timer.timer_id) - - # TODO: Repeating timers should not be removed - self._timer_remove.append(timer.timer_id) + ip, port = address - def _CallSignalWaiters(self, signum): - """Calls all signal waiters for a certain signal. + self.handle_datagram(payload, ip, port) - @type signum: int - @param signum: Signal number + def handle_datagram(self, payload, ip, port): + """Handle an already read udp datagram """ - for owner in self._signal_wait: - owner.OnSignal(signal.SIGCHLD) + raise NotImplementedError + + # this method is overriding an asyncore.dispatcher method + def writable(self): + # We should check whether we can write to the socket only if we have + # something scheduled to be written + return bool(self._out_queue) + + # this method is overriding an asyncore.dispatcher method + def handle_write(self): + if not self._out_queue: + logging.error("handle_write called with empty output queue") + return + (ip, port, payload) = self._out_queue[0] + utils.IgnoreSignals(self.sendto, payload, 0, (ip, port)) + self._out_queue.pop(0) + + def enqueue_send(self, ip, port, payload): + """Enqueue a datagram to be sent when possible - def RegisterIO(self, owner, fd, condition): - """Registers a receiver for I/O notifications + """ + if len(payload) > constants.MAX_UDP_DATA_SIZE: + raise errors.UdpDataSizeError('Packet too big: %s > %s' % (len(payload), + constants.MAX_UDP_DATA_SIZE)) + self._out_queue.append((ip, port, payload)) - The receiver must support a "OnIO(self, fd, conditions)" function. + def process_next_packet(self, timeout=0): + """Process the next datagram, waiting for it if necessary. - @type owner: instance - @param owner: Receiver - @type fd: int - @param fd: File descriptor - @type condition: int - @param condition: ORed field of conditions to be notified - (see select module) + @type timeout: float + @param timeout: how long to wait for data + @rtype: boolean + @return: True if some data has been handled, False otherwise """ - # select.Poller also supports file() like objects, but we don't. - assert isinstance(fd, (int, long)), \ - "Only integers are supported for file descriptors" + result = utils.WaitForFdCondition(self, select.POLLIN, timeout) + if result is not None and result & select.POLLIN: + self.handle_read() + return True + else: + return False + + +class AsyncAwaker(GanetiBaseAsyncoreDispatcher): + """A way to notify the asyncore loop that something is going on. - self._io_wait_add.append((owner, fd, condition)) + If an asyncore daemon is multithreaded when a thread tries to push some data + to a socket, the main loop handling asynchronous requests might be sleeping + waiting on a select(). To avoid this it can create an instance of the + AsyncAwaker, which other threads can use to wake it up. - def UnregisterIO(self, fd): - """Unregister a file descriptor. + """ + def __init__(self, signal_fn=None): + """Constructor for AsyncAwaker - It'll be unregistered the next time the mainloop checks for it. + @type signal_fn: function + @param signal_fn: function to call when awaken - @type fd: int - @param fd: File descriptor + """ + GanetiBaseAsyncoreDispatcher.__init__(self) + assert signal_fn == None or callable(signal_fn) + (self.in_socket, self.out_socket) = socket.socketpair(socket.AF_UNIX, + socket.SOCK_STREAM) + self.in_socket.setblocking(0) + self.in_socket.shutdown(socket.SHUT_WR) + self.out_socket.shutdown(socket.SHUT_RD) + self.set_socket(self.in_socket) + self.need_signal = True + self.signal_fn = signal_fn + self.connected = True + + # this method is overriding an asyncore.dispatcher method + def handle_read(self): + utils.IgnoreSignals(self.recv, 4096) + if self.signal_fn: + self.signal_fn() + self.need_signal = True + + # this method is overriding an asyncore.dispatcher method + def close(self): + asyncore.dispatcher.close(self) + self.out_socket.close() + + def signal(self): + """Signal the asyncore main loop. + + Any data we send here will be ignored, but it will cause the select() call + to return. """ - # select.Poller also supports file() like objects, but we don't. - assert isinstance(fd, (int, long)), \ - "Only integers are supported for file descriptors" + # Yes, there is a race condition here. No, we don't care, at worst we're + # sending more than one wakeup token, which doesn't harm at all. + if self.need_signal: + self.need_signal = False + self.out_socket.send("\0") - self._io_wait_remove.append(fd) + +class Mainloop(object): + """Generic mainloop for daemons + + @ivar scheduler: A sched.scheduler object, which can be used to register + timed events + + """ + def __init__(self): + """Constructs a new Mainloop instance. + + """ + self._signal_wait = [] + self.scheduler = AsyncoreScheduler(time.time) + + @utils.SignalHandled([signal.SIGCHLD]) + @utils.SignalHandled([signal.SIGTERM]) + @utils.SignalHandled([signal.SIGINT]) + def Run(self, signal_handlers=None): + """Runs the mainloop. + + @type signal_handlers: dict + @param signal_handlers: signal->L{utils.SignalHandler} passed by decorator + + """ + assert isinstance(signal_handlers, dict) and \ + len(signal_handlers) > 0, \ + "Broken SignalHandled decorator" + running = True + # Start actual main loop + while running: + if not self.scheduler.empty(): + try: + self.scheduler.run() + except SchedulerBreakout: + pass + else: + asyncore.loop(count=1, use_poll=True) + + # Check whether a signal was raised + for sig in signal_handlers: + handler = signal_handlers[sig] + if handler.called: + self._CallSignalWaiters(sig) + running = sig not in (signal.SIGTERM, signal.SIGINT) + handler.Clear() + + def _CallSignalWaiters(self, signum): + """Calls all signal waiters for a certain signal. + + @type signum: int + @param signum: Signal number + + """ + for owner in self._signal_wait: + owner.OnSignal(signum) def RegisterSignal(self, owner): """Registers a receiver for signal notifications @@ -265,60 +487,88 @@ class Mainloop(object): """ self._signal_wait.append(owner) - def AddTimer(self, owner, interval, repeat): - """Add a new timer. - The receiver must support a "OnTimer(self, timer_id)" function. +def _VerifyDaemonUser(daemon_name): + """Verifies the process uid matches the configured uid. - @type owner: instance - @param owner: Receiver - @type interval: int or float - @param interval: Timer interval in seconds - @type repeat: bool - @param repeat: Whether this is a repeating timer or one-off + This method verifies that a daemon is started as the user it is + intended to be run - """ - # TODO: Implement repeating timers - assert not repeat, "Repeating timers are not yet supported" + @param daemon_name: The name of daemon to be started + @return: A tuple with the first item indicating success or not, + the second item current uid and third with expected uid - # Get new ID - self._timer_id_last += 1 + """ + getents = runtime.GetEnts() + running_uid = os.getuid() + daemon_uids = { + constants.MASTERD: getents.masterd_uid, + constants.RAPI: getents.rapi_uid, + constants.NODED: getents.noded_uid, + constants.CONFD: getents.confd_uid, + } - timer_id = self._timer_id_last + return (daemon_uids[daemon_name] == running_uid, running_uid, + daemon_uids[daemon_name]) - self._timer_add.append(Timer(owner, timer_id, time.time(), - float(interval), repeat)) - return timer_id +def _BeautifyError(err): + """Try to format an error better. - def RemoveTimer(self, timer_id): - """Removes a timer. + Since we're dealing with daemon startup errors, in many cases this + will be due to socket error and such, so we try to format these cases better. - @type timer_id: int - @param timer_id: Timer ID + @param err: an exception object + @rtype: string + @return: the formatted error description - """ - self._timer_remove.append(timer_id) + """ + try: + if isinstance(err, socket.error): + return "Socket-related error: %s (errno=%s)" % (err.args[1], err.args[0]) + elif isinstance(err, EnvironmentError): + if err.filename is None: + return "%s (errno=%s)" % (err.strerror, err.errno) + else: + return "%s (file %s) (errno=%s)" % (err.strerror, err.filename, + err.errno) + else: + return str(err) + except Exception: # pylint: disable-msg=W0703 + logging.exception("Error while handling existing error %s", err) + return "%s" % str(err) -def GenericMain(daemon_name, optionparser, dirs, check_fn, exec_fn): +def GenericMain(daemon_name, optionparser, + check_fn, prepare_fn, exec_fn, + multithreaded=False, console_logging=False, + default_ssl_cert=None, default_ssl_key=None): """Shared main function for daemons. @type daemon_name: string @param daemon_name: daemon name - @type optionparser: L{optparse.OptionParser} + @type optionparser: optparse.OptionParser @param optionparser: initialized optionparser with daemon-specific options (common -f -d options will be handled by this module) - @type options: object @param options: OptionParser result, should contain at - least the fork and the debug options - @type dirs: list of strings - @param dirs: list of directories that must exist for this daemon to work @type check_fn: function which accepts (options, args) @param check_fn: function that checks start conditions and exits if they're not met - @type exec_fn: function which accepts (options, args) + @type prepare_fn: function which accepts (options, args) + @param prepare_fn: function that is run before forking, or None; + it's result will be passed as the third parameter to exec_fn, or + if None was passed in, we will just pass None to exec_fn + @type exec_fn: function which accepts (options, args, prepare_results) @param exec_fn: function that's executed with the daemon's pid file held, and runs the daemon itself. + @type multithreaded: bool + @param multithreaded: Whether the daemon uses threads + @type console_logging: boolean + @param console_logging: if True, the daemon will fall back to the system + console if logging fails + @type default_ssl_cert: string + @param default_ssl_cert: Default SSL certificate path + @type default_ssl_key: string + @param default_ssl_key: Default SSL key path """ optionparser.add_option("-f", "--foreground", dest="fork", @@ -327,37 +577,108 @@ def GenericMain(daemon_name, optionparser, dirs, check_fn, exec_fn): optionparser.add_option("-d", "--debug", dest="debug", help="Enable some debug messages", default=False, action="store_true") + optionparser.add_option("--syslog", dest="syslog", + help="Enable logging to syslog (except debug" + " messages); one of 'no', 'yes' or 'only' [%s]" % + constants.SYSLOG_USAGE, + default=constants.SYSLOG_USAGE, + choices=["no", "yes", "only"]) + if daemon_name in constants.DAEMONS_PORTS: - # for networked daemons we also allow choosing the bind port and address. - # by default we use the port provided by utils.GetDaemonPort, and bind to - # 0.0.0.0 (which is represented by and empty bind address. - port = utils.GetDaemonPort(daemon_name) + default_bind_address = constants.IP4_ADDRESS_ANY + family = ssconf.SimpleStore().GetPrimaryIPFamily() + # family will default to AF_INET if there is no ssconf file (e.g. when + # upgrading a cluster from 2.2 -> 2.3. This is intended, as Ganeti clusters + # <= 2.2 can not be AF_INET6 + if family == netutils.IP6Address.family: + default_bind_address = constants.IP6_ADDRESS_ANY + + default_port = netutils.GetDaemonPort(daemon_name) + + # For networked daemons we allow choosing the port and bind address optionparser.add_option("-p", "--port", dest="port", - help="Network port (%s default)." % port, - default=port, type="int") + help="Network port (default: %s)" % default_port, + default=default_port, type="int") optionparser.add_option("-b", "--bind", dest="bind_address", - help="Bind address", - default="", metavar="ADDRESS") - - multithread = utils.no_fork = daemon_name in constants.MULTITHREADED_DAEMONS + help=("Bind address (default: '%s')" % + default_bind_address), + default=default_bind_address, metavar="ADDRESS") + + if default_ssl_key is not None and default_ssl_cert is not None: + optionparser.add_option("--no-ssl", dest="ssl", + help="Do not secure HTTP protocol with SSL", + default=True, action="store_false") + optionparser.add_option("-K", "--ssl-key", dest="ssl_key", + help=("SSL key path (default: %s)" % + default_ssl_key), + default=default_ssl_key, type="string", + metavar="SSL_KEY_PATH") + optionparser.add_option("-C", "--ssl-cert", dest="ssl_cert", + help=("SSL certificate path (default: %s)" % + default_ssl_cert), + default=default_ssl_cert, type="string", + metavar="SSL_CERT_PATH") + + # Disable the use of fork(2) if the daemon uses threads + utils.no_fork = multithreaded options, args = optionparser.parse_args() - check_fn(options, args) - utils.EnsureDirs(dirs) + if getattr(options, "ssl", False): + ssl_paths = { + "certificate": options.ssl_cert, + "key": options.ssl_key, + } + + for name, path in ssl_paths.iteritems(): + if not os.path.isfile(path): + print >> sys.stderr, "SSL %s file '%s' was not found" % (name, path) + sys.exit(constants.EXIT_FAILURE) + + # TODO: By initiating http.HttpSslParams here we would only read the files + # once and have a proper validation (isfile returns False on directories) + # at the same time. + + result, running_uid, expected_uid = _VerifyDaemonUser(daemon_name) + if not result: + msg = ("%s started using wrong user ID (%d), expected %d" % + (daemon_name, running_uid, expected_uid)) + print >> sys.stderr, msg + sys.exit(constants.EXIT_FAILURE) + + if check_fn is not None: + check_fn(options, args) if options.fork: utils.CloseFDs() - utils.Daemonize(logfile=constants.DAEMONS_LOGFILES[daemon_name]) + wpipe = utils.Daemonize(logfile=constants.DAEMONS_LOGFILES[daemon_name]) + else: + wpipe = None - utils.WritePidFile(daemon_name) + utils.WritePidFile(utils.DaemonPidFileName(daemon_name)) try: - utils.SetupLogging(logfile=constants.DAEMONS_LOGFILES[daemon_name], - debug=options.debug, - stderr_logging=not options.fork, - multithreaded=multithread) - logging.info("%s daemon startup" % daemon_name) - exec_fn(options, args) + try: + utils.SetupLogging(logfile=constants.DAEMONS_LOGFILES[daemon_name], + debug=options.debug, + stderr_logging=not options.fork, + multithreaded=multithreaded, + program=daemon_name, + syslog=options.syslog, + console_logging=console_logging) + if callable(prepare_fn): + prep_results = prepare_fn(options, args) + else: + prep_results = None + logging.info("%s daemon startup", daemon_name) + except Exception, err: + utils.WriteErrorToFD(wpipe, _BeautifyError(err)) + raise + + if wpipe is not None: + # we're done with the preparation phase, we close the pipe to + # let the parent know it's safe to exit + os.close(wpipe) + + exec_fn(options, args, prep_results) finally: utils.RemovePidFile(daemon_name) -