/ - Diff - Synnefo - Greek Research and Technology Network's projects

Revision 659de616

b/snf-cyclades-app/synnefo/app_settings/default/queues.py
9	9	AMQP_BACKEND = 'puka'
10	10
11	11	EXCHANGE_GANETI = "ganeti" # Messages from Ganeti
12		EXCHANGE_CRON = "cron" # Messages from periodically triggered tasks
13		EXCHANGE_API = "api" # Messages from the REST API
14		EXCHANGES = (EXCHANGE_GANETI, EXCHANGE_CRON, EXCHANGE_API)

     from synnefo.lib.amqp import AMQPClient
     from synnefo.logic import callbacks
     from synnefo.logic import queues
     import logging
     log = logging.getLogger()
     # Queue names
     QUEUES = []
     # Queue bindings to exchanges
     BINDINGS = []
     class Dispatcher:
         debug = False
-...
             self.client.close()
         def _init(self):
             global QUEUES, BINDINGS
             log.info("Initializing")
             self.client = AMQPClient()
-...
             self.client.connect()
             # Declare queues and exchanges
             for exchange in settings.EXCHANGES:
                 self.client.exchange_declare(exchange=exchange,
                                              type="topic")
             for queue in QUEUES:
             exchange = settings.EXCHANGE_GANETI
             exchange_dl = queues.convert_exchange_to_dead(exchange)
             self.client.exchange_declare(exchange=exchange,
                                          type="topic")
             self.client.exchange_declare(exchange=exchange_dl,
                                          type="topic")
             for queue in queues.QUEUES:
                 # Queues are mirrored to all RabbitMQ brokers
                 self.client.queue_declare(queue=queue, mirrored=True)
             bindings = BINDINGS
                 self.client.queue_declare(queue=queue, mirrored=True,
                                           dead_letter_exchange=exchange_dl)
                 # Declare the corresponding dead-letter queue
                 queue_dl = queues.convert_queue_to_dead(queue)
                 self.client.queue_declare(queue=queue_dl, mirrored=True)
             # Bind queues to handler methods
             for binding in bindings:
             for binding in queues.BINDINGS:
                 try:
                     callback = getattr(callbacks, binding[3])
                 except AttributeError:
                     log.error("Cannot find callback %s", binding[3])
                     raise SystemExit(1)
                 queue = binding[0]
                 exchange = binding[1]
                 routing_key = binding[2]
                 self.client.queue_bind(queue=binding[0], exchange=binding[1],
                                        routing_key=binding[2])
                 self.client.queue_bind(queue=queue, exchange=exchange,
                                        routing_key=routing_key)
                 self.client.basic_consume(queue=binding[0],
                                           callback=callback,
                                           prefetch_count=5)
                 queue_dl = queues.convert_queue_to_dead(queue)
                 exchange_dl = queues.convert_exchange_to_dead(exchange)
                 # Bind the corresponding dead-letter queue
                 self.client.queue_bind(queue=queue_dl,
                                        exchange=exchange_dl,
                                        routing_key=routing_key)
                 log.debug("Binding %s(%s) to queue %s with handler %s",
                           binding[1], binding[2], binding[0], binding[3])
     def _init_queues():
         global QUEUES, BINDINGS
         # Queue declarations
         prefix = settings.BACKEND_PREFIX_ID.split('-')[0]
         QUEUE_GANETI_EVENTS_OP = "%s-events-op" % prefix
         QUEUE_GANETI_EVENTS_NETWORK = "%s-events-network" % prefix
         QUEUE_GANETI_EVENTS_NET = "%s-events-net" % prefix
         QUEUE_GANETI_BUILD_PROGR = "%s-events-progress" % prefix
         QUEUE_RECONC = "%s-reconciliation" % prefix
         if settings.DEBUG is True:
             QUEUE_DEBUG = "%s-debug" % prefix  # Debug queue, retrieves all messages
         QUEUES = (QUEUE_GANETI_EVENTS_OP, QUEUE_GANETI_EVENTS_NETWORK, QUEUE_GANETI_EVENTS_NET, QUEUE_RECONC,
                   QUEUE_GANETI_BUILD_PROGR)
         # notifications of type "ganeti-op-status"
         DB_HANDLER_KEY_OP = 'ganeti.%s.event.op' % prefix
         # notifications of type "ganeti-network-status"
         DB_HANDLER_KEY_NETWORK = 'ganeti.%s.event.network' % prefix
         # notifications of type "ganeti-net-status"
         DB_HANDLER_KEY_NET = 'ganeti.%s.event.net' % prefix
         # notifications of type "ganeti-create-progress"
         BUILD_MONITOR_HANDLER = 'ganeti.%s.event.progress' % prefix
         # reconciliation
         RECONC_HANDLER = 'reconciliation.%s.*' % prefix
         BINDINGS = [
         # Queue                   # Exchange                # RouteKey              # Handler
         (QUEUE_GANETI_EVENTS_OP,  settings.EXCHANGE_GANETI, DB_HANDLER_KEY_OP,      'update_db'),
         (QUEUE_GANETI_EVENTS_NETWORK, settings.EXCHANGE_GANETI, DB_HANDLER_KEY_NETWORK, 'update_network'),
         (QUEUE_GANETI_EVENTS_NET, settings.EXCHANGE_GANETI, DB_HANDLER_KEY_NET,     'update_net'),
         (QUEUE_GANETI_BUILD_PROGR, settings.EXCHANGE_GANETI, BUILD_MONITOR_HANDLER,  'update_build_progress'),
+        ]
         if settings.DEBUG is True:
             BINDINGS += [
                 # Queue       # Exchange          # RouteKey  # Handler
                 (QUEUE_DEBUG, settings.EXCHANGE_GANETI, '#',  'dummy_proc'),
                 (QUEUE_DEBUG, settings.EXCHANGE_CRON,   '#',  'dummy_proc'),
                 (QUEUE_DEBUG, settings.EXCHANGE_API,    '#',  'dummy_proc'),
+            ]
             QUEUES += (QUEUE_DEBUG,)
                           exchange, routing_key, queue, binding[3])
     def parse_arguments(args):
-...
         """
             Delete declared queues from RabbitMQ. Use with care!
         """
         global QUEUES, BINDINGS
         client = AMQPClient(max_retries=120)
         client.connect()
         print "Queues to be deleted: ", QUEUES
         print "Queues to be deleted: ", queues.QUEUES
         if not get_user_confirmation():
             return
         for queue in QUEUES:
         for queue in queues.QUEUES:
             result = client.queue_delete(queue=queue)
             print "Deleting queue %s. Result: %s" % (queue, result)
-...
     def purge_exchanges():
         """Delete declared exchanges from RabbitMQ, after removing all queues"""
         global QUEUES, BINDINGS
         purge_queues()
         client = AMQPClient()
         client.connect()
         print "Exchanges to be deleted: ", settings.EXCHANGES
         exchanges = queues.EXCHANGES
         print "Exchanges to be deleted: ", exchanges
         if not get_user_confirmation():
             return
         for exchange in settings.EXCHANGES:
             result = client.exchange_delete(exchange=exchange)
             print "Deleting exchange %s. Result: %s" % (exchange, result)
         for exch in exchanges:
             result = client.exchange_delete(exchange=exch)
             print "Deleting exchange %s. Result: %s" % (exch, result)
         client.close()
     def drain_queue(queue):
         """Strip a (declared) queue from all outstanding messages"""
         global QUEUES, BINDINGS
         if not queue:
             return
         if not queue in QUEUES:
         if not queue in queues.QUEUES:
             print "Queue %s not configured" % queue
             return
-...
         setproctitle.setproctitle(sys.argv[0])
         setup_logging(opts)
         # Init the global variables containing the queues
         _init_queues()
         # Special case for the clean up queues action
         if opts.purge_queues:
             purge_queues()

     # Copyright 2012 GRNET S.A. All rights reserved.
+    #
     # Redistribution and use in source and binary forms, with or without
     # modification, are permitted provided that the following conditions
     # are met:
+    #
     #   1. Redistributions of source code must retain the above copyright
     #      notice, this list of conditions and the following disclaimer.
+    #
     #  2. Redistributions in binary form must reproduce the above copyright
     #     notice, this list of conditions and the following disclaimer in the
     #     documentation and/or other materials provided with the distribution.
+    #
     # THIS SOFTWARE IS PROVIDED BY THE REGENTS AND CONTRIBUTORS ``AS IS'' AND
     # ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
     # IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
     # ARE DISCLAIMED.  IN NO EVENT SHALL THE REGENTS OR CONTRIBUTORS BE LIABLE
     # FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
     # DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
     # OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
     # HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
     # LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
     # OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
     # SUCH DAMAGE.
+    #
     # The views and conclusions contained in the software and documentation are
     # those of the authors and should not be interpreted as representing official
     # policies, either expressed or implied, of GRNET S.A.
     from synnefo.settings import BACKEND_PREFIX_ID, DEBUG, EXCHANGE_GANETI
     try:
         prefix = BACKEND_PREFIX_ID.split('-')[0]
     except TypeError, IndexError:
         raise Exception("Invalid BACKEND_PREFIX_ID")
     # EXCHANGES
     EXCHANGES = (EXCHANGE_GANETI,)
     # QUEUES
     QUEUE_OP = "%s-events-op" % prefix
     QUEUE_NETWORK = "%s-events-network" % prefix
     QUEUE_NET = "%s-events-net" % prefix
     QUEUE_PROGRESS = "%s-events-progress" % prefix
     QUEUES = (QUEUE_OP,
               QUEUE_NETWORK,
               QUEUE_NET,
               QUEUE_PROGRESS)
     # ROUTING KEYS
     # notifications of type "ganeti-op-status"
     KEY_OP = 'ganeti.%s.event.op' % prefix
     # notifications of type "ganeti-network-status"
     KEY_NETWORK = 'ganeti.%s.event.network' % prefix
     # notifications of type "ganeti-net-status"
     KEY_NET = 'ganeti.%s.event.net' % prefix
     # notifications of type "ganeti-create-progress"
     KEY_PROGRESS = 'ganeti.%s.event.progress' % prefix
     # BINDINGS:
     BINDINGS = (
     # Queue             # Exchange          # RouteKey    # Handler
     (QUEUE_OP,          EXCHANGE_GANETI,    KEY_OP,       'update_db'),
     (QUEUE_NETWORK,     EXCHANGE_GANETI,    KEY_NETWORK,  'update_network'),
     (QUEUE_NET,         EXCHANGE_GANETI,    KEY_NET,      'update_net'),
     (QUEUE_PROGRESS,    EXCHANGE_GANETI,    KEY_PROGRESS, 'update_build_progress'),
+    )
     ## Extra for DEBUG:
     if DEBUG is True:
         # Debug queue, retrieves all messages
         QUEUE_DEBUG = "%s-debug" % prefix
         QUEUES += (QUEUE_DEBUG,)
         BINDINGS += ((QUEUE_DEBUG, EXCHANGE_GANETI, "#", "dummy_proc"),)
     def convert_queue_to_dead(queue):
         """Convert the name of a queue to the corresponding dead-letter one"""
         return queue + "-dl"
     def convert_exchange_to_dead(exchange):
         """Convert the name of an exchange to the corresponding dead-letter one"""
         return exchange + "-dl"

Also available in: Unified diff