Fix AMQP bug in snf-progress-monitor

snf-cyclades-gtools/synnefo/ganeti/eventd.py

@@ -119,7 +119,7 @@ class JobFileHandler(pyinotify.ProcessEvent):
         self.logger = logger
         self.cluster_name = cluster_name
 
-        self.client = AMQPClient(confirm_buffer=25)
+        self.client = AMQPClient(hosts=settings.AMQP_HOSTS, confirm_buffer=25)
         handler_logger.info("Attempting to connect to RabbitMQ hosts")
 
         self.client.connect()

snf-cyclades-gtools/synnefo/ganeti/hook.py

@@ -164,7 +164,8 @@ class GanetiHook():
 
         # FIXME: We need a reconciliation mechanism between the DB and
         #        Ganeti, for cases exactly like this.
-        self.client = AMQPClient(max_retries=2 * len(settings.AMQP_HOSTS))
+        self.client = AMQPClient(hosts=settings.AMQP_HOSTS,
+                                 max_retries = 2 * len(settings.AMQP_HOSTS))
         self.client.connect()
 
     def on_master(self):

snf-cyclades-gtools/synnefo/ganeti/progress_monitor.py

@@ -130,8 +130,9 @@ def main():
     # determine the routekey for AMPQ
     prefix = opts.instance_name.split('-')[0]
     routekey = "ganeti.%s.event.progress" % prefix
-    amqp_client = AMQPClient()
+    amqp_client = AMQPClient(hosts=settings.AMQP_HOSTS, confirm_buffer=2)
     amqp_client.connect()
+    amqp_client.exchange_declare(settings.EXCHANGE_GANETI, type='topic')
 
     pid = os.fork()
     if pid == 0:
@@ -202,6 +203,8 @@ def main():
         # Sleep for a while
         time.sleep(3)
 
+    amqp_client.close()
+
 
 if __name__ == "__main__":
     sys.exit(main())