root/fedd/fedd.py

#!/usr/bin/env python

import os,sys

from optparse import OptionParser

from federation import config_parser
from federation.server import server, xmlrpc_handler, soap_handler
from federation.util import fedd_ssl_context, file_expanding_opts
from federation.deter_impl import new_feddservice

from socket import error as socket_error
from threading import Lock, Thread
from signal import signal, pause, SIGINT, SIGTERM
from select import select, error
from time import sleep

import logging
import M2Crypto
import os.path

class fedd_opts(file_expanding_opts):
    """Encapsulate option processing in this class, rather than in main"""

    def __init__(self):
        file_expanding_opts.__init__(self, 
                usage="%prog [opts] (--help for details)",
                version="0.1")

        self.set_defaults(logfile=None, debug=0)

        self.add_option("-d", "--debug", action="count", dest="debug", 
                help="Set debug.  Repeat for more information")
        self.add_option("-f", "--configfile", action="callback",
                callback=self.expand_file, type='str',
                default="/usr/local/etc/fedd.conf",
                dest="configfile", help="Configuration file")
        self.add_option("-l", "--logfile", action="store", dest="logfile", 
                help="File to send log messages to")
        self.add_option("--trace", action="store_const", dest="tracefile", 
                const=sys.stderr, help="Print SOAP exchange to stderr")

servers_active = True       # Sub-servers run while this is True
servers = [ ]               # server instances instantiated from services
servers_lock = Lock()       # Lock to manipulate servers from sub-server threads

def shutdown(sig, frame):
    """
    On a signal, stop running sub-servers.  
    
    This is connected to signals below
    """
    global servers_active, flog

    servers_active = False
    flog.info("Received signal %d, shutting down" % sig);

def run_server(s):
    """
    Operate a subserver, shutting down when servers_active is false.

    Each server (that is host/port/transport triple) has a thread running this
    function, so each can handle requests independently.  They all call in to
    the same implementation, which must manage its own synchronization.
    """
    global servers_active   # Not strictly needed: servers_active is only read
    global servers          # List of active servers
    global servers_lock     # Lock to manipulate servers

    while servers_active:
        try:
            i, o, e = select((s,), (), (), 1.0)
            if s in i: s.handle_request()
        except error:
            # The select call seems to get interrupted by signals as well as
            # the main thread.  This essentially ignores signals in this
            # thread.
            pass

    # Done.  Remove us from the list
    servers_lock.acquire()
    servers.remove(s)
    servers_lock.release()
M2Crypto.threading.init()
opts, args = fedd_opts().parse_args()

# Logging setup
flog = logging.getLogger("fedd")
ffmt = logging.Formatter("%(asctime)s %(name)s %(message)s",
        '%d %b %y %H:%M:%S')

if opts.logfile: fh = logging.FileHandler(opts.logfile)
else: fh = logging.StreamHandler(sys.stdout)

# The handler will print anything, setting the logger level will affect what
# gets recorded.
fh.setLevel(logging.DEBUG)

if opts.debug: flog.setLevel(logging.DEBUG)
else: flog.setLevel(logging.INFO)

fh.setFormatter(ffmt)
flog.addHandler(fh)


if not os.access(opts.configfile, os.R_OK):
    sys.exit("Can't read config file %s" % opts.configfile)

# Initialize the implementation
try:
    config= config_parser()
    config.read(opts.configfile)
except Exception, e:
    sys.exit("Cannot parse config file %s: %s" % (opts.configfile, e))

try:
    impl = new_feddservice(config)
except RuntimeError, e:
    str = getattr(e, 'desc', None) or getattr(e,'message', None) or \
            "No message"
    sys.exit("Error configuring fedd: %s" % str)

if impl.cert_file:
    if not os.access(impl.cert_file, os.R_OK):
        sys.exit("Cannot read certificate file: %s" % impl.cert_file)
else:
    sys.exit("Must supply certificate file (probably in config)")

if impl.trusted_certs:
    if not os.access(impl.trusted_certs, os.R_OK):
        sys.exit("Cannot read trusted certificate file: %s" % \
                impl.trusted_certs)

# Create the SSL credentials
ctx = None
while ctx == None:
    try:
        ctx = fedd_ssl_context(impl.cert_file, impl.trusted_certs, 
                password=impl.cert_pwd)
    except Exception, e:
        if str(e) != "bad decrypt" or impl.cert_pwd != None:
            raise

services = config.get("globals", "services", "23235")

for s in services.split(","):
    s = s.strip()
    colons = s.count(":")
    try:
        if colons == 0:
            p = int(s)
            h = ''
            t = 'soap'
        elif colons == 1:
            p, t  = s.split(":")
            p = int(p)
            h = ''
        elif colons == 2:
            h, p, t  = s.split(":")
            p = int(p)
        else:
            flog.error("Invalid service specification %s ignored." % s)
            continue
    except ValueError:
        flog.error("Error converting port to integer in %s: spec ignored" % s)
        continue

    t = t.lower()
    sdebug = (opts.debug > 0)
    try:
        if t == 'soap':
            servers.append(server((h, p), soap_handler, ctx, impl, sdebug))
        elif t == 'xmlrpc':
            servers.append(server((h, p), xmlrpc_handler, ctx, impl, sdebug))
        else:
            flog.error("Invalid transport specification (%s) in service %s" % \
                    (t, s))
            continue
    except socket_error, e:
        flog.error("Cannot create server for %s: %s" % (s, e[1]))
        continue

#  Make sure that there are no malformed servers in the list
servers = [ s for s in servers if s ]

# Catch signals
signal(SIGINT, shutdown)
signal(SIGTERM, shutdown)

# Start the servers
for s in servers:
    Thread(target=run_server, args=(s,)).start()

# Main thread waits for signals
while servers_active:
    sleep(1.0)

#Once shutdown starts wait for all the servers to terminate.
while True:
    servers_lock.acquire()
    if len(servers) == 0: 
        servers_lock.release()
        flog.info("All servers exited.  Terminating")
        sys.exit(0)
    servers_lock.release()
    sleep(1)