Merge "Amphora Create now fails when amphora goes into ERROR status"

octavia/controller/worker/flows/amphora_flows.py

@@ -16,7 +16,6 @@
 from oslo_config import cfg
 from taskflow.patterns import graph_flow
 from taskflow.patterns import linear_flow
-from taskflow import retry
 
 from octavia.common import constants
 from octavia.controller.worker.tasks import amphora_driver_tasks
@@ -64,17 +63,12 @@ class AmphoraFlows(object):
                 provides=constants.COMPUTE_ID))
         create_amphora_flow.add(database_tasks.MarkAmphoraBootingInDB(
             requires=(constants.AMPHORA_ID, constants.COMPUTE_ID)))
-        wait_flow = linear_flow.Flow(constants.WAIT_FOR_AMPHORA,
+        create_amphora_flow.add(compute_tasks.ComputeWait(
-                                     retry=retry.Times(CONF.
-                                                       controller_worker.
-                                                       amp_active_retries))
-        wait_flow.add(compute_tasks.ComputeWait(
             requires=constants.COMPUTE_ID,
             provides=constants.COMPUTE_OBJ))
-        wait_flow.add(database_tasks.UpdateAmphoraInfo(
+        create_amphora_flow.add(database_tasks.UpdateAmphoraInfo(
             requires=(constants.AMPHORA_ID, constants.COMPUTE_OBJ),
             provides=constants.AMPHORA))
-        create_amphora_flow.add(wait_flow)
         create_amphora_flow.add(database_tasks.ReloadAmphora(
             requires=constants.AMPHORA_ID,
             provides=constants.AMPHORA))
@@ -148,20 +142,14 @@ class AmphoraFlows(object):
         create_amp_for_lb_subflow.add(database_tasks.MarkAmphoraBootingInDB(
             name=sf_name + '-' + constants.MARK_AMPHORA_BOOTING_INDB,
             requires=(constants.AMPHORA_ID, constants.COMPUTE_ID)))
-        wait_flow = linear_flow.Flow(sf_name + '-' +
+        create_amp_for_lb_subflow.add(compute_tasks.ComputeWait(
-                                     constants.WAIT_FOR_AMPHORA,
-                                     retry=retry.Times(CONF.
-                                                       controller_worker.
-                                                       amp_active_retries))
-        wait_flow.add(compute_tasks.ComputeWait(
             name=sf_name + '-' + constants.COMPUTE_WAIT,
             requires=constants.COMPUTE_ID,
             provides=constants.COMPUTE_OBJ))
-        wait_flow.add(database_tasks.UpdateAmphoraInfo(
+        create_amp_for_lb_subflow.add(database_tasks.UpdateAmphoraInfo(
             name=sf_name + '-' + constants.UPDATE_AMPHORA_INFO,
             requires=(constants.AMPHORA_ID, constants.COMPUTE_OBJ),
             provides=constants.AMPHORA))
-        create_amp_for_lb_subflow.add(wait_flow)
         create_amp_for_lb_subflow.add(amphora_driver_tasks.AmphoraFinalize(
             name=sf_name + '-' + constants.AMPHORA_FINALIZE,
             requires=constants.AMPHORA))
@@ -315,17 +303,12 @@ class AmphoraFlows(object):
                 requires=(constants.AMPHORA_ID, constants.LOADBALANCER_ID)))
         failover_amphora_flow.add(database_tasks.MarkAmphoraBootingInDB(
             requires=(constants.AMPHORA_ID, constants.COMPUTE_ID)))
-        wait_flow = linear_flow.Flow(constants.WAIT_FOR_AMPHORA,
+        failover_amphora_flow.add(compute_tasks.ComputeWait(
-                                     retry=retry.Times(CONF.
-                                                       controller_worker.
-                                                       amp_active_retries))
-        wait_flow.add(compute_tasks.ComputeWait(
             requires=constants.COMPUTE_ID,
             provides=constants.COMPUTE_OBJ))
-        wait_flow.add(database_tasks.UpdateAmphoraInfo(
+        failover_amphora_flow.add(database_tasks.UpdateAmphoraInfo(
             requires=(constants.AMPHORA_ID, constants.COMPUTE_OBJ),
             provides=constants.AMPHORA))
-        failover_amphora_flow.add(wait_flow)
         failover_amphora_flow.add(database_tasks.ReloadAmphora(
             requires=constants.AMPHORA_ID,
             provides=constants.FAILOVER_AMPHORA))

octavia/controller/worker/tasks/compute_tasks.py

@@ -153,9 +153,12 @@ class ComputeWait(BaseComputeTask):
         :raises: Generic exception if the amphora is not active
         :returns: An amphora object
         """
-        time.sleep(CONF.controller_worker.amp_active_wait_sec)
+        for i in range(CONF.controller_worker.amp_active_retries):
-        amp = self.compute.get_amphora(compute_id)
+            amp = self.compute.get_amphora(compute_id)
-        if amp.status == constants.ACTIVE:
+            if amp.status == constants.ACTIVE:
-            return amp
+                return amp
+            elif amp.status == constants.ERROR:
+                raise exceptions.ComputeBuildException()
+            time.sleep(CONF.controller_worker.amp_active_wait_sec)
 
         raise exceptions.ComputeWaitTimeoutException()

octavia/tests/unit/controller/worker/tasks/test_compute_tasks.py

@@ -13,8 +13,6 @@
 # under the License.
 #
 
-import time
-
 import mock
 from oslo_config import cfg
 from oslo_config import fixture as oslo_fixture
@@ -255,8 +253,6 @@ class TestComputeTasks(base.TestCase):
         computewait = compute_tasks.ComputeWait()
         computewait.execute(COMPUTE_ID)
 
-        time.sleep.assert_called_once_with(AMP_WAIT)
-
         mock_driver.get_amphora.assert_called_once_with(COMPUTE_ID)
 
         _amphora_mock.status = constants.DELETED
@@ -265,6 +261,27 @@ class TestComputeTasks(base.TestCase):
                           computewait.execute,
                           _amphora_mock)
 
+    @mock.patch('stevedore.driver.DriverManager.driver')
+    @mock.patch('time.sleep')
+    def test_compute_wait_error_status(self, mock_time_sleep, mock_driver):
+
+        _amphora_mock.compute_id = COMPUTE_ID
+        _amphora_mock.status = constants.ACTIVE
+        _amphora_mock.lb_network_ip = LB_NET_IP
+
+        mock_driver.get_amphora.return_value = _amphora_mock
+
+        computewait = compute_tasks.ComputeWait()
+        computewait.execute(COMPUTE_ID)
+
+        mock_driver.get_amphora.assert_called_once_with(COMPUTE_ID)
+
+        _amphora_mock.status = constants.ERROR
+
+        self.assertRaises(exceptions.ComputeBuildException,
+                          computewait.execute,
+                          _amphora_mock)
+
     @mock.patch('stevedore.driver.DriverManager.driver')
     def test_delete_amphorae_on_load_balancer(self, mock_driver):