code.grnet.gr Git - ganeti-local/blob - lib/daemon.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 """Module with helper classes and functions for daemons"""
  23
  24
  25 import asyncore
  26 import asynchat
  27 import collections
  28 import grp
  29 import os
  30 import pwd
  31 import signal
  32 import logging
  33 import sched
  34 import time
  35 import socket
  36 import select
  37 import sys
  38
  39 from ganeti import utils
  40 from ganeti import constants
  41 from ganeti import errors
  42 from ganeti import netutils
  43
  44
  45 _DEFAULT_RUN_USER = "root"
  46 _DEFAULT_RUN_GROUP = "root"
  47
  48
  49 class SchedulerBreakout(Exception):
  50   """Exception used to get out of the scheduler loop
  51
  52   """
  53
  54
  55 def AsyncoreDelayFunction(timeout):
  56   """Asyncore-compatible scheduler delay function.
  57
  58   This is a delay function for sched that, rather than actually sleeping,
  59   executes asyncore events happening in the meantime.
  60
  61   After an event has occurred, rather than returning, it raises a
  62   SchedulerBreakout exception, which will force the current scheduler.run()
  63   invocation to terminate, so that we can also check for signals. The main loop
  64   will then call the scheduler run again, which will allow it to actually
  65   process any due events.
  66
  67   This is needed because scheduler.run() doesn't support a count=..., as
  68   asyncore loop, and the scheduler module documents throwing exceptions from
  69   inside the delay function as an allowed usage model.
  70
  71   """
  72   asyncore.loop(timeout=timeout, count=1, use_poll=True)
  73   raise SchedulerBreakout()
  74
  75
  76 class AsyncoreScheduler(sched.scheduler):
  77   """Event scheduler integrated with asyncore
  78
  79   """
  80   def __init__(self, timefunc):
  81     sched.scheduler.__init__(self, timefunc, AsyncoreDelayFunction)
  82
  83
  84 class GanetiBaseAsyncoreDispatcher(asyncore.dispatcher):
  85   """Base Ganeti Asyncore Dispacher
  86
  87   """
  88   # this method is overriding an asyncore.dispatcher method
  89   def handle_error(self):
  90     """Log an error in handling any request, and proceed.
  91
  92     """
  93     logging.exception("Error while handling asyncore request")
  94
  95   # this method is overriding an asyncore.dispatcher method
  96   def writable(self):
  97     """Most of the time we don't want to check for writability.
  98
  99     """
 100     return False
 101
 102
 103 def FormatAddress(family, address):
 104   """Format a client's address
 105
 106   @type family: integer
 107   @param family: socket family (one of socket.AF_*)
 108   @type address: family specific (usually tuple)
 109   @param address: address, as reported by this class
 110
 111   """
 112   if family == socket.AF_INET and len(address) == 2:
 113     return "%s:%d" % address
 114   elif family == socket.AF_UNIX and len(address) == 3:
 115     return "pid=%s, uid=%s, gid=%s" % address
 116   else:
 117     return str(address)
 118
 119
 120 class AsyncStreamServer(GanetiBaseAsyncoreDispatcher):
 121   """A stream server to use with asyncore.
 122
 123   Each request is accepted, and then dispatched to a separate asyncore
 124   dispatcher to handle.
 125
 126   """
 127
 128   _REQUEST_QUEUE_SIZE = 5
 129
 130   def __init__(self, family, address):
 131     """Constructor for AsyncUnixStreamSocket
 132
 133     @type family: integer
 134     @param family: socket family (one of socket.AF_*)
 135     @type address: address family dependent
 136     @param address: address to bind the socket to
 137
 138     """
 139     GanetiBaseAsyncoreDispatcher.__init__(self)
 140     self.family = family
 141     self.create_socket(self.family, socket.SOCK_STREAM)
 142     self.set_reuse_addr()
 143     self.bind(address)
 144     self.listen(self._REQUEST_QUEUE_SIZE)
 145
 146   # this method is overriding an asyncore.dispatcher method
 147   def handle_accept(self):
 148     """Accept a new client connection.
 149
 150     Creates a new instance of the handler class, which will use asyncore to
 151     serve the client.
 152
 153     """
 154     accept_result = utils.IgnoreSignals(self.accept)
 155     if accept_result is not None:
 156       connected_socket, client_address = accept_result
 157       if self.family == socket.AF_UNIX:
 158         # override the client address, as for unix sockets nothing meaningful
 159         # is passed in from accept anyway
 160         client_address = netutils.GetSocketCredentials(connected_socket)
 161       logging.info("Accepted connection from %s",
 162                    FormatAddress(self.family, client_address))
 163       self.handle_connection(connected_socket, client_address)
 164
 165   def handle_connection(self, connected_socket, client_address):
 166     """Handle an already accepted connection.
 167
 168     """
 169     raise NotImplementedError
 170
 171
 172 class AsyncTerminatedMessageStream(asynchat.async_chat):
 173   """A terminator separated message stream asyncore module.
 174
 175   Handles a stream connection receiving messages terminated by a defined
 176   separator. For each complete message handle_message is called.
 177
 178   """
 179   def __init__(self, connected_socket, peer_address, terminator, family,
 180                unhandled_limit):
 181     """AsyncTerminatedMessageStream constructor.
 182
 183     @type connected_socket: socket.socket
 184     @param connected_socket: connected stream socket to receive messages from
 185     @param peer_address: family-specific peer address
 186     @type terminator: string
 187     @param terminator: terminator separating messages in the stream
 188     @type family: integer
 189     @param family: socket family
 190     @type unhandled_limit: integer or None
 191     @param unhandled_limit: maximum unanswered messages
 192
 193     """
 194     # python 2.4/2.5 uses conn=... while 2.6 has sock=... we have to cheat by
 195     # using a positional argument rather than a keyword one.
 196     asynchat.async_chat.__init__(self, connected_socket)
 197     self.connected_socket = connected_socket
 198     # on python 2.4 there is no "family" attribute for the socket class
 199     # FIXME: when we move to python 2.5 or above remove the family parameter
 200     #self.family = self.connected_socket.family
 201     self.family = family
 202     self.peer_address = peer_address
 203     self.terminator = terminator
 204     self.unhandled_limit = unhandled_limit
 205     self.set_terminator(terminator)
 206     self.ibuffer = []
 207     self.receive_count = 0
 208     self.send_count = 0
 209     self.oqueue = collections.deque()
 210     self.iqueue = collections.deque()
 211
 212   # this method is overriding an asynchat.async_chat method
 213   def collect_incoming_data(self, data):
 214     self.ibuffer.append(data)
 215
 216   def _can_handle_message(self):
 217     return (self.unhandled_limit is None or
 218             (self.receive_count < self.send_count + self.unhandled_limit) and
 219              not self.iqueue)
 220
 221   # this method is overriding an asynchat.async_chat method
 222   def found_terminator(self):
 223     message = "".join(self.ibuffer)
 224     self.ibuffer = []
 225     message_id = self.receive_count
 226     # We need to increase the receive_count after checking if the message can
 227     # be handled, but before calling handle_message
 228     can_handle = self._can_handle_message()
 229     self.receive_count += 1
 230     if can_handle:
 231       self.handle_message(message, message_id)
 232     else:
 233       self.iqueue.append((message, message_id))
 234
 235   def handle_message(self, message, message_id):
 236     """Handle a terminated message.
 237
 238     @type message: string
 239     @param message: message to handle
 240     @type message_id: integer
 241     @param message_id: stream's message sequence number
 242
 243     """
 244     pass
 245     # TODO: move this method to raise NotImplementedError
 246     # raise NotImplementedError
 247
 248   def send_message(self, message):
 249     """Send a message to the remote peer. This function is thread-safe.
 250
 251     @type message: string
 252     @param message: message to send, without the terminator
 253
 254     @warning: If calling this function from a thread different than the one
 255     performing the main asyncore loop, remember that you have to wake that one
 256     up.
 257
 258     """
 259     # If we just append the message we received to the output queue, this
 260     # function can be safely called by multiple threads at the same time, and
 261     # we don't need locking, since deques are thread safe. handle_write in the
 262     # asyncore thread will handle the next input message if there are any
 263     # enqueued.
 264     self.oqueue.append(message)
 265
 266   # this method is overriding an asyncore.dispatcher method
 267   def readable(self):
 268     # read from the socket if we can handle the next requests
 269     return self._can_handle_message() and asynchat.async_chat.readable(self)
 270
 271   # this method is overriding an asyncore.dispatcher method
 272   def writable(self):
 273     # the output queue may become full just after we called writable. This only
 274     # works if we know we'll have something else waking us up from the select,
 275     # in such case, anyway.
 276     return asynchat.async_chat.writable(self) or self.oqueue
 277
 278   # this method is overriding an asyncore.dispatcher method
 279   def handle_write(self):
 280     if self.oqueue:
 281       # if we have data in the output queue, then send_message was called.
 282       # this means we can process one more message from the input queue, if
 283       # there are any.
 284       data = self.oqueue.popleft()
 285       self.push(data + self.terminator)
 286       self.send_count += 1
 287       if self.iqueue:
 288         self.handle_message(*self.iqueue.popleft())
 289     self.initiate_send()
 290
 291   def close_log(self):
 292     logging.info("Closing connection from %s",
 293                  FormatAddress(self.family, self.peer_address))
 294     self.close()
 295
 296   # this method is overriding an asyncore.dispatcher method
 297   def handle_expt(self):
 298     self.close_log()
 299
 300   # this method is overriding an asyncore.dispatcher method
 301   def handle_error(self):
 302     """Log an error in handling any request, and proceed.
 303
 304     """
 305     logging.exception("Error while handling asyncore request")
 306     self.close_log()
 307
 308
 309 class AsyncUDPSocket(GanetiBaseAsyncoreDispatcher):
 310   """An improved asyncore udp socket.
 311
 312   """
 313   def __init__(self, family):
 314     """Constructor for AsyncUDPSocket
 315
 316     """
 317     GanetiBaseAsyncoreDispatcher.__init__(self)
 318     self._out_queue = []
 319     self._family = family
 320     self.create_socket(family, socket.SOCK_DGRAM)
 321
 322   # this method is overriding an asyncore.dispatcher method
 323   def handle_connect(self):
 324     # Python thinks that the first udp message from a source qualifies as a
 325     # "connect" and further ones are part of the same connection. We beg to
 326     # differ and treat all messages equally.
 327     pass
 328
 329   # this method is overriding an asyncore.dispatcher method
 330   def handle_read(self):
 331     recv_result = utils.IgnoreSignals(self.recvfrom,
 332                                       constants.MAX_UDP_DATA_SIZE)
 333     if recv_result is not None:
 334       payload, address = recv_result
 335       if self._family == socket.AF_INET6:
 336         # we ignore 'flow info' and 'scope id' as we don't need them
 337         ip, port, _, _ = address
 338       else:
 339         ip, port = address
 340
 341       self.handle_datagram(payload, ip, port)
 342
 343   def handle_datagram(self, payload, ip, port):
 344     """Handle an already read udp datagram
 345
 346     """
 347     raise NotImplementedError
 348
 349   # this method is overriding an asyncore.dispatcher method
 350   def writable(self):
 351     # We should check whether we can write to the socket only if we have
 352     # something scheduled to be written
 353     return bool(self._out_queue)
 354
 355   # this method is overriding an asyncore.dispatcher method
 356   def handle_write(self):
 357     if not self._out_queue:
 358       logging.error("handle_write called with empty output queue")
 359       return
 360     (ip, port, payload) = self._out_queue[0]
 361     utils.IgnoreSignals(self.sendto, payload, 0, (ip, port))
 362     self._out_queue.pop(0)
 363
 364   def enqueue_send(self, ip, port, payload):
 365     """Enqueue a datagram to be sent when possible
 366
 367     """
 368     if len(payload) > constants.MAX_UDP_DATA_SIZE:
 369       raise errors.UdpDataSizeError('Packet too big: %s > %s' % (len(payload),
 370                                     constants.MAX_UDP_DATA_SIZE))
 371     self._out_queue.append((ip, port, payload))
 372
 373   def process_next_packet(self, timeout=0):
 374     """Process the next datagram, waiting for it if necessary.
 375
 376     @type timeout: float
 377     @param timeout: how long to wait for data
 378     @rtype: boolean
 379     @return: True if some data has been handled, False otherwise
 380
 381     """
 382     result = utils.WaitForFdCondition(self, select.POLLIN, timeout)
 383     if result is not None and result & select.POLLIN:
 384       self.handle_read()
 385       return True
 386     else:
 387       return False
 388
 389
 390 class AsyncAwaker(GanetiBaseAsyncoreDispatcher):
 391   """A way to notify the asyncore loop that something is going on.
 392
 393   If an asyncore daemon is multithreaded when a thread tries to push some data
 394   to a socket, the main loop handling asynchronous requests might be sleeping
 395   waiting on a select(). To avoid this it can create an instance of the
 396   AsyncAwaker, which other threads can use to wake it up.
 397
 398   """
 399   def __init__(self, signal_fn=None):
 400     """Constructor for AsyncAwaker
 401
 402     @type signal_fn: function
 403     @param signal_fn: function to call when awaken
 404
 405     """
 406     GanetiBaseAsyncoreDispatcher.__init__(self)
 407     assert signal_fn == None or callable(signal_fn)
 408     (self.in_socket, self.out_socket) = socket.socketpair(socket.AF_UNIX,
 409                                                           socket.SOCK_STREAM)
 410     self.in_socket.setblocking(0)
 411     self.in_socket.shutdown(socket.SHUT_WR)
 412     self.out_socket.shutdown(socket.SHUT_RD)
 413     self.set_socket(self.in_socket)
 414     self.need_signal = True
 415     self.signal_fn = signal_fn
 416     self.connected = True
 417
 418   # this method is overriding an asyncore.dispatcher method
 419   def handle_read(self):
 420     utils.IgnoreSignals(self.recv, 4096)
 421     if self.signal_fn:
 422       self.signal_fn()
 423     self.need_signal = True
 424
 425   # this method is overriding an asyncore.dispatcher method
 426   def close(self):
 427     asyncore.dispatcher.close(self)
 428     self.out_socket.close()
 429
 430   def signal(self):
 431     """Signal the asyncore main loop.
 432
 433     Any data we send here will be ignored, but it will cause the select() call
 434     to return.
 435
 436     """
 437     # Yes, there is a race condition here. No, we don't care, at worst we're
 438     # sending more than one wakeup token, which doesn't harm at all.
 439     if self.need_signal:
 440       self.need_signal = False
 441       self.out_socket.send("\0")
 442
 443
 444 class Mainloop(object):
 445   """Generic mainloop for daemons
 446
 447   @ivar scheduler: A sched.scheduler object, which can be used to register
 448     timed events
 449
 450   """
 451   def __init__(self):
 452     """Constructs a new Mainloop instance.
 453
 454     """
 455     self._signal_wait = []
 456     self.scheduler = AsyncoreScheduler(time.time)
 457
 458   @utils.SignalHandled([signal.SIGCHLD])
 459   @utils.SignalHandled([signal.SIGTERM])
 460   @utils.SignalHandled([signal.SIGINT])
 461   def Run(self, signal_handlers=None):
 462     """Runs the mainloop.
 463
 464     @type signal_handlers: dict
 465     @param signal_handlers: signal->L{utils.SignalHandler} passed by decorator
 466
 467     """
 468     assert isinstance(signal_handlers, dict) and \
 469            len(signal_handlers) > 0, \
 470            "Broken SignalHandled decorator"
 471     running = True
 472     # Start actual main loop
 473     while running:
 474       if not self.scheduler.empty():
 475         try:
 476           self.scheduler.run()
 477         except SchedulerBreakout:
 478           pass
 479       else:
 480         asyncore.loop(count=1, use_poll=True)
 481
 482       # Check whether a signal was raised
 483       for sig in signal_handlers:
 484         handler = signal_handlers[sig]
 485         if handler.called:
 486           self._CallSignalWaiters(sig)
 487           running = sig not in (signal.SIGTERM, signal.SIGINT)
 488           handler.Clear()
 489
 490   def _CallSignalWaiters(self, signum):
 491     """Calls all signal waiters for a certain signal.
 492
 493     @type signum: int
 494     @param signum: Signal number
 495
 496     """
 497     for owner in self._signal_wait:
 498       owner.OnSignal(signum)
 499
 500   def RegisterSignal(self, owner):
 501     """Registers a receiver for signal notifications
 502
 503     The receiver must support a "OnSignal(self, signum)" function.
 504
 505     @type owner: instance
 506     @param owner: Receiver
 507
 508     """
 509     self._signal_wait.append(owner)
 510
 511
 512 def GenericMain(daemon_name, optionparser, dirs, check_fn, exec_fn,
 513                 multithreaded=False, console_logging=False,
 514                 default_ssl_cert=None, default_ssl_key=None,
 515                 user=_DEFAULT_RUN_USER, group=_DEFAULT_RUN_GROUP):
 516   """Shared main function for daemons.
 517
 518   @type daemon_name: string
 519   @param daemon_name: daemon name
 520   @type optionparser: optparse.OptionParser
 521   @param optionparser: initialized optionparser with daemon-specific options
 522                        (common -f -d options will be handled by this module)
 523   @type dirs: list of (string, integer)
 524   @param dirs: list of directories that must be created if they don't exist,
 525                and the permissions to be used to create them
 526   @type check_fn: function which accepts (options, args)
 527   @param check_fn: function that checks start conditions and exits if they're
 528                    not met
 529   @type exec_fn: function which accepts (options, args)
 530   @param exec_fn: function that's executed with the daemon's pid file held, and
 531                   runs the daemon itself.
 532   @type multithreaded: bool
 533   @param multithreaded: Whether the daemon uses threads
 534   @type console_logging: boolean
 535   @param console_logging: if True, the daemon will fall back to the system
 536                           console if logging fails
 537   @type default_ssl_cert: string
 538   @param default_ssl_cert: Default SSL certificate path
 539   @type default_ssl_key: string
 540   @param default_ssl_key: Default SSL key path
 541   @param user: Default user to run as
 542   @type user: string
 543   @param group: Default group to run as
 544   @type group: string
 545
 546   """
 547   optionparser.add_option("-f", "--foreground", dest="fork",
 548                           help="Don't detach from the current terminal",
 549                           default=True, action="store_false")
 550   optionparser.add_option("-d", "--debug", dest="debug",
 551                           help="Enable some debug messages",
 552                           default=False, action="store_true")
 553   optionparser.add_option("--syslog", dest="syslog",
 554                           help="Enable logging to syslog (except debug"
 555                           " messages); one of 'no', 'yes' or 'only' [%s]" %
 556                           constants.SYSLOG_USAGE,
 557                           default=constants.SYSLOG_USAGE,
 558                           choices=["no", "yes", "only"])
 559
 560   if daemon_name in constants.DAEMONS_PORTS:
 561     default_bind_address = constants.IP4_ADDRESS_ANY
 562     default_port = netutils.GetDaemonPort(daemon_name)
 563
 564     # For networked daemons we allow choosing the port and bind address
 565     optionparser.add_option("-p", "--port", dest="port",
 566                             help="Network port (default: %s)" % default_port,
 567                             default=default_port, type="int")
 568     optionparser.add_option("-b", "--bind", dest="bind_address",
 569                             help=("Bind address (default: %s)" %
 570                                   default_bind_address),
 571                             default=default_bind_address, metavar="ADDRESS")
 572
 573   if default_ssl_key is not None and default_ssl_cert is not None:
 574     optionparser.add_option("--no-ssl", dest="ssl",
 575                             help="Do not secure HTTP protocol with SSL",
 576                             default=True, action="store_false")
 577     optionparser.add_option("-K", "--ssl-key", dest="ssl_key",
 578                             help=("SSL key path (default: %s)" %
 579                                   default_ssl_key),
 580                             default=default_ssl_key, type="string",
 581                             metavar="SSL_KEY_PATH")
 582     optionparser.add_option("-C", "--ssl-cert", dest="ssl_cert",
 583                             help=("SSL certificate path (default: %s)" %
 584                                   default_ssl_cert),
 585                             default=default_ssl_cert, type="string",
 586                             metavar="SSL_CERT_PATH")
 587
 588   # Disable the use of fork(2) if the daemon uses threads
 589   utils.no_fork = multithreaded
 590
 591   options, args = optionparser.parse_args()
 592
 593   if getattr(options, "ssl", False):
 594     ssl_paths = {
 595       "certificate": options.ssl_cert,
 596       "key": options.ssl_key,
 597       }
 598
 599     for name, path in ssl_paths.iteritems():
 600       if not os.path.isfile(path):
 601         print >> sys.stderr, "SSL %s file '%s' was not found" % (name, path)
 602         sys.exit(constants.EXIT_FAILURE)
 603
 604     # TODO: By initiating http.HttpSslParams here we would only read the files
 605     # once and have a proper validation (isfile returns False on directories)
 606     # at the same time.
 607
 608   if check_fn is not None:
 609     check_fn(options, args)
 610
 611   utils.EnsureDirs(dirs)
 612
 613   if options.fork:
 614     try:
 615       uid = pwd.getpwnam(user).pw_uid
 616       gid = grp.getgrnam(group).gr_gid
 617     except KeyError:
 618       raise errors.ConfigurationError("User or group not existing on system:"
 619                                       " %s:%s" % (user, group))
 620     utils.CloseFDs()
 621     utils.Daemonize(constants.DAEMONS_LOGFILES[daemon_name], uid, gid)
 622
 623   utils.WritePidFile(daemon_name)
 624   try:
 625     utils.SetupLogging(logfile=constants.DAEMONS_LOGFILES[daemon_name],
 626                        debug=options.debug,
 627                        stderr_logging=not options.fork,
 628                        multithreaded=multithreaded,
 629                        program=daemon_name,
 630                        syslog=options.syslog,
 631                        console_logging=console_logging)
 632     logging.info("%s daemon startup", daemon_name)
 633     exec_fn(options, args)
 634   finally:
 635     utils.RemovePidFile(daemon_name)