import os
import sys
import logging
-import asyncore
-import socket
-import pyinotify
import time
+try:
+ from pyinotify import pyinotify
+except ImportError:
+ import pyinotify
+
from optparse import OptionParser
+from ganeti import asyncnotifier
+from ganeti import confd
+from ganeti.confd import server as confd_server
from ganeti import constants
from ganeti import errors
from ganeti import daemon
from ganeti import ssconf
-from ganeti.asyncnotifier import AsyncNotifier
-from ganeti.confd.server import ConfdProcessor
-class ConfdAsyncUDPServer(asyncore.dispatcher):
+class ConfdAsyncUDPServer(daemon.AsyncUDPSocket):
"""The confd udp server, suitable for use with asyncore.
"""
@type port: int
@param port: udp port
@type processor: L{confd.server.ConfdProcessor}
- @param reader: ConfigReader to use to access the config
+ @param processor: ConfdProcessor to use to handle queries
"""
- asyncore.dispatcher.__init__(self)
+ daemon.AsyncUDPSocket.__init__(self)
self.bind_address = bind_address
self.port = port
self.processor = processor
- self.create_socket(socket.AF_INET, socket.SOCK_DGRAM)
self.bind((bind_address, port))
logging.debug("listening on ('%s':%d)" % (bind_address, port))
- # 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):
+ # this method is overriding a daemon.AsyncUDPSocket method
+ def handle_datagram(self, payload_in, ip, port):
try:
- payload_in, address = self.recvfrom(4096)
- ip, port = address
- payload_out = self.processor.ExecQuery(payload_in, ip, port)
- if payload_out is not None:
- self.sendto(payload_out, 0, (ip, port))
- except:
- # we need to catch any exception here, log it, but proceed, because even
- # if we failed handling a single request, we still want the confd to
- # continue working.
- logging.error("Unexpected exception", exc_info=True)
+ query = confd.UnpackMagic(payload_in)
+ except errors.ConfdMagicError, err:
+ logging.debug(err)
+ return
- # this method is overriding an asyncore.dispatcher method
- def writable(self):
- # No need to check if we can write to the UDP socket
- return False
+ answer = self.processor.ExecQuery(query, ip, port)
+ if answer is not None:
+ try:
+ self.enqueue_send(ip, port, confd.PackMagic(answer))
+ except errors.UdpDataSizeError:
+ logging.error("Reply too big to fit in an udp packet.")
class ConfdInotifyEventHandler(pyinotify.ProcessEvent):
pyinotify.EventsCodes.IN_MODIFY
self.file = file
self.watch_handle = None
- self.enable()
def enable(self):
"""Watch the given file
check, to verify that the reload hasn't failed.
"""
- def __init__(self, reader, mainloop):
+ def __init__(self, processor, mainloop):
"""Constructor for ConfdConfigurationReloader
- @type reader: L{ssconf.SimpleConfigReader}
- @param reader: ganeti-confd SimpleConfigReader
+ @type processor: L{confd.server.ConfdProcessor}
+ @param processor: ganeti-confd ConfdProcessor
@type mainloop: L{daemon.Mainloop}
@param mainloop: ganeti-confd mainloop
"""
- self.reader = reader
+ self.processor = processor
self.mainloop = mainloop
- self.polling = False
+ self.polling = True
self.last_notification = 0
# Asyncronous inotify handler for config changes
self.wm = pyinotify.WatchManager()
self.inotify_handler = ConfdInotifyEventHandler(self.wm, self.OnInotify)
- self.notifier = AsyncNotifier(self.wm, self.inotify_handler)
+ self.notifier = asyncnotifier.AsyncNotifier(self.wm, self.inotify_handler)
self.timer_handle = None
self._EnableTimer()
try:
self.inotify_handler.enable()
except errors.InotifyError:
- raise errors.ConfdFatalError(err)
+ self.polling = True
try:
- reloaded = self.reader.Reload()
+ reloaded = self.processor.reader.Reload()
if reloaded:
logging.info("Reloaded ganeti config")
else:
logging.debug("Skipped double config reload")
except errors.ConfigurationError:
- # transform a ConfigurationError in a fatal error, that will cause confd
- # to quit.
- raise errors.ConfdFatalError(err)
+ self.DisableConfd()
+ self.inotify_handler.disable()
+ return
# Reset the timer. If we're polling it will go to the polling rate, if
# we're not it will delay it again to its base safe timeout.
- self._DisableTimer()
- self._EnableTimer()
+ self._ResetTimer()
def _DisableTimer(self):
if self.timer_handle is not None:
self.timer_handle = self.mainloop.scheduler.enter(
timeout, 1, self.OnTimer, [])
+ def _ResetTimer(self):
+ self._DisableTimer()
+ self._EnableTimer()
+
def OnTimer(self):
"""Function called when the timer fires
"""
self.timer_handle = None
+ reloaded = False
+ was_disabled = False
try:
- reloaded = self.reader.Reload()
+ if self.processor.reader is None:
+ was_disabled = True
+ self.EnableConfd()
+ reloaded = True
+ else:
+ reloaded = self.processor.reader.Reload()
except errors.ConfigurationError:
- # transform a ConfigurationError in a fatal error, that will cause confd
- # to quit.
- raise errors.ConfdFatalError(err)
+ self.DisableConfd(silent=was_disabled)
+ return
if self.polling and reloaded:
logging.info("Reloaded ganeti config")
# We have reloaded the config files, but received no inotify event. If
# an event is pending though, we just happen to have timed out before
# receiving it, so this is not a problem, and we shouldn't alert
- if not self.notifier.check_events():
+ if not self.notifier.check_events() and not was_disabled:
logging.warning("Config file reload at timeout (inotify failure)")
elif self.polling:
# We're polling, but we haven't reloaded the config:
# Going back to inotify mode
logging.debug("Moving from polling mode to inotify mode")
self.polling = False
- self.inotify_handler.enable()
+ try:
+ self.inotify_handler.enable()
+ except errors.InotifyError:
+ self.polling = True
else:
logging.debug("Performed configuration check")
self._EnableTimer()
+ def DisableConfd(self, silent=False):
+ """Puts confd in non-serving mode
+
+ """
+ if not silent:
+ logging.warning("Confd is being disabled")
+ self.processor.Disable()
+ self.polling = False
+ self._ResetTimer()
+
+ def EnableConfd(self):
+ self.processor.Enable()
+ logging.warning("Confd is being enabled")
+ self.polling = True
+ self._ResetTimer()
+
def CheckConfd(options, args):
"""Initial checks whether to run exit with a failure.
print >> sys.stderr, "Need HMAC key %s to run" % constants.HMAC_CLUSTER_KEY
sys.exit(constants.EXIT_FAILURE)
- ssconf.CheckMasterCandidate(options.debug)
-
def ExecConfd(options, args):
"""Main confd function, executed with PID file held
"""
mainloop = daemon.Mainloop()
- # confd-level SimpleConfigReader
- reader = ssconf.SimpleConfigReader()
-
# Asyncronous confd UDP server
- processor = ConfdProcessor(reader)
+ processor = confd_server.ConfdProcessor()
+ try:
+ processor.Enable()
+ except errors.ConfigurationError:
+ # If enabling the processor has failed, we can still go on, but confd will
+ # be disabled
+ logging.warning("Confd is starting in disabled mode")
+ pass
server = ConfdAsyncUDPServer(options.bind_address, options.port, processor)
# Configuration reloader
- reloader = ConfdConfigurationReloader(reader, mainloop)
+ reloader = ConfdConfigurationReloader(processor, mainloop)
mainloop.Run()
constants.RELEASE_VERSION)
dirs = [(val, constants.RUN_DIRS_MODE) for val in constants.SUB_RUN_DIRS]
- dirs.append((constants.LOG_OS_DIR, 0750))
dirs.append((constants.LOCK_DIR, 1777))
daemon.GenericMain(constants.CONFD, parser, dirs, CheckConfd, ExecConfd)
projects / ganeti-local / blobdiff