Producer/endpoint code to allow for amphora failovers

Change-Id: I06b0ec785610c2113384e1da638a56e93b1ab38d

octavia/api/handlers/queue/producer.py

@@ -120,6 +120,28 @@ class LoadBalancerProducer(BaseProducer):
         self.client.cast({}, method_name, **kw)
 
 
+class AmphoraProducer(BaseProducer):
+    """Sends failover messages to the RPC end of the queue consumer
+
+    """
+    PAYLOAD_CLASS = "amphora"
+
+    @property
+    def payload_class(self):
+        return self.PAYLOAD_CLASS
+
+    def failover(self, data_model):
+        """sends a failover message to the controller via oslo.messaging
+
+        :param data_model:
+        """
+        model_id = getattr(data_model, 'id', None)
+        p_class = self.payload_class
+        kw = {"{0}_id".format(p_class): model_id}
+        method_name = "failover_{0}".format(self.payload_class)
+        self.client.cast({}, method_name, **kw)
+
+
 class ListenerProducer(BaseProducer):
     """Sends updates,deletes and creates to the RPC end of the queue consumer
 
@@ -215,3 +237,4 @@ class ProducerHandler(abstract_handler.BaseHandler):
     member = MemberProducer()
     l7policy = L7PolicyProducer()
     l7rule = L7RuleProducer()
+    amphora = AmphoraProducer()

octavia/controller/queue/endpoint.py

@@ -58,6 +58,11 @@ class Endpoint(object):
                  load_balancer_id)
         self.worker.failover_loadbalancer(load_balancer_id)
 
+    def failover_amphora(self, context, amphora_id):
+        LOG.info('Failing over amphora \'%s\'...',
+                 amphora_id)
+        self.worker.failover_amphora(amphora_id)
+
     def create_listener(self, context, listener_id):
         LOG.info('Creating listener \'%s\'...', listener_id)
         self.worker.create_listener(listener_id)

octavia/tests/unit/api/handlers/queue/test_producer.py

@@ -78,6 +78,20 @@ class TestProducer(base.TestCase):
         self.mck_client.cast.assert_called_once_with(
             {}, 'delete_load_balancer', **kw)
 
+    def test_failover_loadbalancer(self):
+        p = producer.LoadBalancerProducer()
+        p.failover(self.mck_model)
+        kw = {'load_balancer_id': self.mck_model.id}
+        self.mck_client.cast.assert_called_once_with(
+            {}, 'failover_load_balancer', **kw)
+
+    def test_failover_amphora(self):
+        p = producer.AmphoraProducer()
+        p.failover(self.mck_model)
+        kw = {'amphora_id': self.mck_model.id}
+        self.mck_client.cast.assert_called_once_with(
+            {}, 'failover_amphora', **kw)
+
     def test_update_loadbalancer(self):
         p = producer.LoadBalancerProducer()
         lb = data_models.LoadBalancer(id=10)

octavia/tests/unit/controller/queue/test_endpoint.py

@@ -61,6 +61,11 @@ class TestEndpoint(base.TestCase):
         self.ep.worker.failover_loadbalancer.assert_called_once_with(
             self.resource_id)
 
+    def test_failover_amphora(self):
+        self.ep.failover_amphora(self.context, self.resource_id)
+        self.ep.worker.failover_amphora.assert_called_once_with(
+            self.resource_id)
+
     def test_create_listener(self):
         self.ep.create_listener(self.context, self.resource_id)
         self.ep.worker.create_listener.assert_called_once_with(