Merge lp:~larry-e-works/uci-engine/amqp-to-kombu into lp:uci-engine

[Larry E Works]

Changed the bits in ci-utils/ci_utils/amqp_utils.py to use kombu instead of libamqp; has not been fully tested yet as I receive an error when attempting to run run-tests on a bootstack deployment. Will wipe that out and try a local LXC deployment.

Merged from trunk to gather in all of the changes since work started on this; no conflicts.

Changed the bits in ci-utils/ci_utils/amqp_utils.py to use kombu instead of libamqp; has not been fully tested yet as I receive an error when attempting to run run-tests on a bootstack deployment. Will wipe that out and try a local LXC deployment.

[lp:~larry-e-works/uci-engine/amqp-to-kombu]

770. By Larry E Works on 2014-10-01

Seems I missed a comma at the end of the password=rabbit_config.AMQP_PASSWORD line. Have added and will re-run.

771. By Larry E Works on 2014-10-01

Merged from trunk while I was at it; no conflicts.

[Larry E Works (larry-e-works)]

Was thinking about this earlier (and the traceback received when attempting to run the tests) and the problem originates with with a missing comma in setting the kombu connection string. Have fixed that and will try again.

[lp:~larry-e-works/uci-engine/amqp-to-kombu]

772. By Larry E Works on 2014-10-02

Modified send() in amqp_utils.pu to call kombu in the correct manner. This worked with the test in the sense that I can ssh into the rabbit server and see the test queue get created and populated. The rest of the test still fails since the test itself (tests.test_rabbit.TestRabbit.test_acked_message) is still written to use libamqp functionality that doesn not directly translate to kombu functionality.

Modified venv.py to add the needed kombu dependencies for testing. The deps have been downloaded locally, added and pushed to lp:~canonical-ci-engineering/uci-engine/deps/

Unmerged revisions

772. By Larry E Works on 2014-10-02

Modified send() in amqp_utils.pu to call kombu in the correct manner. This worked with the test in the sense that I can ssh into the rabbit server and see the test queue get created and populated. The rest of the test still fails since the test itself (tests.test_rabbit.TestRabbit.test_acked_message) is still written to use libamqp functionality that doesn not directly translate to kombu functionality.

Modified venv.py to add the needed kombu dependencies for testing. The deps have been downloaded locally, added and pushed to lp:~canonical-ci-engineering/uci-engine/deps/

771. By Larry E Works on 2014-10-01

Merged from trunk while I was at it; no conflicts.

770. By Larry E Works on 2014-10-01

Seems I missed a comma at the end of the password=rabbit_config.AMQP_PASSWORD line. Have added and will re-run.

769. By Larry E Works on 2014-09-30

Massive merge from trunk, no conflicts.

768. By Larry E Works on 2014-09-08

Initial try at replacing amqp with kombu.

=== modified file 'ci-utils/ci_utils/amqp_utils.py'
--- ci-utils/ci_utils/amqp_utils.py	2014-08-19 15:10:19 +0000
+++ ci-utils/ci_utils/amqp_utils.py	2014-10-02 20:36:00 +0000
@@ -19,7 +19,9 @@
 import socket
 import time
 
-from amqplib import client_0_8 as amqp
+##TOAD##from amqplib import client_0_8 as amqp
+from kombu import Connection, Producer, Exchange, Queue
+from kombu.mixins import ConsumerMixin
 
 BSBUILDER_QUEUE = 'bsbuilder'
 IMAGE_BUILDER_QUEUE = 'imagebuilder'
@@ -46,6 +48,17 @@
 HERE = os.path.abspath(os.path.dirname(__file__))
 
 
+class Worker(ConsumerMixin):
+
+    def __init__(self, connection, queue, callback):
+        self.connection = connection
+        self.queue = queue
+        return
+
+    def get_consumers(self, Consumer, channel):
+        return [Consumer(self.queue, callbacks=callback)]
+
+
 def get_config():
     '''Load the rabbit config created by the charm'''
     config = None
@@ -65,11 +78,12 @@
 
 
 def connection(rabbit_config):
-    return amqp.Connection(
+    return Connection(
+        'librabbitmq://' + rabbit_config.AMQP_HOST + '//',
         userid=rabbit_config.AMQP_USER,
+        password=rabbit_config.AMQP_PASSWORD,
         virtual_host=rabbit_config.AMQP_VHOST,
-        host=rabbit_config.AMQP_HOST,
-        password=rabbit_config.AMQP_PASSWORD
+        failover_strategy='round-robin'
     )
 
 
@@ -79,12 +93,12 @@
     conn = channel = None
     try:
         conn = connection(config)
-        channel = conn.channel()
-        channel.queue_declare(queue=queue_name, durable=True,
-                              auto_delete=False)
+        conn.ensure_connection()
+        print "  Connected: " + str(conn.connection)
+        queue = Queue(queue_name, durable=True, auto_delete=False)
+        channel = queue(conn)
+        channel.declare()
     except Exception as e:
-        if channel:
-            channel.close()
         if conn:
             conn.close()
         raise e
@@ -104,17 +118,17 @@
     conn = channel = None
     try:
         conn, channel = declare_queue(queue_name, config)
-        body = amqp.Message(msg)
-        body.properties['delivery_mode'] = 2  # Persistent
-        channel.basic_publish(body, exchange='', routing_key=queue_name)
+        producer = Producer(conn)
+        producer.publish(msg, routing_key=queue_name, delivery_mode=2)
+##TOAD##        body = amqp.Message(msg)
+##TOAD##        body.properties['delivery_mode'] = 2  # Persistent
+##TOAD##        channel.basic_publish(body, exchange='', routing_key=queue_name)
     except Exception as e:
         if raise_errors:
             raise
         logging.exception('unable to queue up message: %s', e)
         return str(e)
     finally:
-        if channel:
-            channel.close()
         if conn:
             conn.close()
     return None
@@ -122,14 +136,14 @@
 
 def _run_forever(channel, queue, callback, retry_period=120):
     logging.info('Waiting for messages. ^C to exit.')
-    tag = channel.basic_consume(callback=callback, queue=queue)
+    tag = channel.ConsumerMixin.Consumer(queue, callbacks=callback)
     try:
         timeout = time.time()
         while channel.callbacks and time.time() < timeout + retry_period:
             try:
                 channel.wait()
                 timeout = time.time()
-            except (amqp.AMQPConnectionException, socket.error):
+            except (kombu.ConnectionError, socket.error):
                 logging.error('lost connection to Rabbit')
                 # TODO metrics.meter('lost_rabbit_connection')
                 # Don't probe immediately, give the network/process

=== modified file 'testing/venv.py'
--- testing/venv.py	2014-09-04 17:02:00 +0000
+++ testing/venv.py	2014-10-02 20:36:00 +0000
@@ -107,7 +107,8 @@
 
 def additional_deps():
     '''Install addtional dependencies.'''
-    additional_deps = ['cheetah', 'python-apt', 'jinja2']
+    additional_deps = ['cheetah', 'python-apt', 'jinja2', 'kombu', 'amqp',
+                       'librabbitmq', 'anyjson']
     for dep in additional_deps:
         cache_file = 'file://%s' % os.environ['PIP_CACHE']
         cmd = ['pip', 'install', '--no-index', '-f', cache_file, dep]