Synnefo

class Dispatcher:

    logger = None

    chan = None

    def __init__(self, debug, logger):

        self.logger = logger

        self._init_queues(debug)

    def update_db(self, message):

        try:

            msg = json.loads(message.body)

            if msg["type"] != "ganeti-op-status":

                self.logger.error("Message is of uknown type %s." % (msg["type"],))

                return

            vmid = utils.id_from_instance_name(msg["instance"])

            vm = VirtualMachine.objects.get(id=vmid)

            self.logger.debug("Processing msg: %s" % (msg,))

            backend.process_backend_msg(vm, msg["jobId"], msg["operation"], msg["status"], msg["logmsg"])

            self.logger.debug("Done processing msg for vm %s." % (msg["instance"]))

        except KeyError:

            self.logger.error("Malformed incoming JSON, missing attributes: " + message.body)

        except VirtualMachine.InvalidBackendIdError:

            self.logger.debug("Ignoring msg for unknown instance %s." % (msg["instance"],))

        except VirtualMachine.DoesNotExist:

            self.logger.error("VM for instance %s with id %d not found in DB." % (msg["instance"], vmid))

        except Exception as e:

            self.logger.error("Unexpected error:\n" + "".join(traceback.format_exception(*sys.exc_info())))

        finally:

            message.channel.basic_ack(message.delivery_tag)

    def send_email(self, message):

        self.logger.debug("Request to send email message")

    def update_credits(self, message):

        self.logger.debug("Request to update credits")

        message.channel.basic_ack(message.delivery_tag)

    def wait(self):

        while True:

            try:

                self.chan.wait()

            except SystemExit:

                break

        self.chan.basic_cancel("dbupdater")

        self.chan.close()

        self.chan.connection.close()

    def _declare_queues(self):

        self.chan.exchange_declare(exchange=settings.EXCHANGE_GANETI, type="direct", durable=True, auto_delete=False)

        self.chan.exchange_declare(exchange=settings.EXCHANGE_CRON, type="topic", durable=True, auto_delete=False)

        self.chan.exchange_declare(exchange=settings.EXCHANGE_API, type="topic", durable=True, auto_delete=False)

        self.chan.queue_declare(queue=settings.QUEUE_GANETI_EVENTS, durable=True, exclusive=False, auto_delete=False)

        self.chan.queue_declare(queue=settings.QUEUE_CRON_CREDITS, durable=True, exclusive=False, auto_delete=False)

        self.chan.queue_declare(queue=settings.QUEUE_API_EMAIL, durable=True, exclusive=False, auto_delete=False)

        self.chan.queue_declare(queue=settings.QUEUE_CRON_EMAIL, durable=True, exclusive=False, auto_delete=False)

    def _init_queues(self,debug):

        self._open_channel()

        if debug:

            self._init_devel()

        else:

            self._init()

    def _init_devel(self):

        self._declare_queues()

        self.chan.queue_bind(queue=settings.QUEUE_GANETI_EVENTS, exchange=settings.EXCHANGE_GANETI, routing_key="event.*")

        self.chan.basic_consume(queue="events", callback=self.update_db, consumer_tag="dbupdater")

    def _init(self):

        self._declare_queues()

        self.chan.queue_bind(queue=settings.QUEUE_GANETI_EVENTS, exchange=settings.EXCHANGE_GANETI, routing_key="event.*")

        self.chan.basic_consume(queue="events", callback=self.update_db, consumer_tag="dbupdater")

    def _open_channel(self):

        conn = None

        while conn == None:

            self.logger.info("Attempting to connect to %s", settings.RABBIT_HOST)

            try:

                conn = amqp.Connection( host=settings.RABBIT_HOST,

                                    userid=settings.RABBIT_USERNAME,

                                    password=settings.RABBIT_PASSWORD,

                                    virtual_host=settings.RABBIT_VHOST)

            except socket.error:

                time.sleep(1)

                pass

        self.logger.info("Connection succesful, opening channel")

        self.chan = conn.channel()

def child(cmdline):

    #Cmd line argument parsing

    (opts, args) = parse_arguments(cmdline)

    d = Dispatcher(debug = True, logger = logger)

    d.wait()

def parse_arguments(args):

    from optparse import OptionParser

    parser = OptionParser()

    parser.add_option("-d", "--debug", action="store_true", dest="debug",

            help="Enable debug mode")

    parser.add_option("-l", "--log", dest="log_file",

            default=settings.DISPATCHER_LOG_FILE,

            metavar="FILE",

            help="Write log to FILE instead of %s" %

            settings.DISPATCHER_LOG_FILE)

    return parser.parse_args(args)

def main():

    global logger

    (opts, args) = parse_arguments(sys.argv[1:])

    #newpid = os.fork()

    #if newpid == 0:

    child(sys.argv[1:])

    #else:

    #    pids = (os.getpid(), newpid)

    #    print "parent: %d, child: %d" % pids

    #daemon_context = daemon.DaemonContext(

    #        umask=022,

    #        stdout=handler.stream,

    #        stderr=handler.stream,

    #        files_preserve=[handler.stream])

    #daemon_context.open()

    #logger.info("Became a daemon")

    #signal(SIGINT, exit_handler)

    #signal(SIGTERM, exit_handler)

    #signal(SIGKILL, exit_handler)