Merge "Fix member operating status when add/remove HM"

octavia/controller/worker/flows/health_monitor_flows.py

@@ -68,6 +68,10 @@ class HealthMonitorFlows(object):
             requires=constants.HEALTH_MON))
         delete_hm_flow.add(database_tasks.DecrementHealthMonitorQuota(
             requires=constants.HEALTH_MON))
+        delete_hm_flow.add(
+            database_tasks.UpdatePoolMembersOperatingStatusInDB(
+                requires=constants.POOL,
+                inject={constants.OPERATING_STATUS: constants.NO_MONITOR}))
         delete_hm_flow.add(database_tasks.MarkPoolActiveInDB(
             requires=constants.POOL))
         delete_hm_flow.add(database_tasks.MarkLBAndListenersActiveInDB(

octavia/controller/worker/tasks/database_tasks.py

@@ -2643,3 +2643,25 @@ class CountPoolChildrenForQuota(BaseDatabaseTask):
         member_count = len(pool.members)
 
         return {'HM': health_mon_count, 'member': member_count}
+
+
+class UpdatePoolMembersOperatingStatusInDB(BaseDatabaseTask):
+    """Updates the members of a pool operating status.
+
+    Since sqlalchemy will likely retry by itself always revert if it fails
+    """
+
+    def execute(self, pool, operating_status):
+        """Update the members of a pool operating status in DB.
+
+        :param pool: Pool object to be updated
+        :param operating_status: Operating status to set
+        :returns: None
+        """
+
+        LOG.debug("Updating member operating status to %(status)s in DB for "
+                  "pool id: %(pool)s", {'status': operating_status,
+                                        'pool': pool.id})
+        self.member_repo.update_pool_members(db_apis.get_session(),
+                                             pool.id,
+                                             operating_status=operating_status)

octavia/db/repositories.py

@@ -746,6 +746,18 @@ class MemberRepository(BaseRepository):
         """Batch deletes members from a pool."""
         self.delete_batch(session, member_ids)
 
+    def update_pool_members(self, session, pool_id, **model_kwargs):
+        """Updates all of the members of a pool.
+
+        :param session: A Sql Alchemy database session.
+        :param pool_id: ID of the pool to update members on.
+        :param model_kwargs: Entity attributes that should be updates.
+        :returns: octavia.common.data_model
+        """
+        with session.begin(subtransactions=True):
+            session.query(self.model_class).filter_by(
+                pool_id=pool_id).update(model_kwargs)
+
 
 class ListenerRepository(BaseRepository):
     model_class = models.Listener

octavia/tests/functional/db/test_repositories.py

@@ -2031,6 +2031,20 @@ class MemberRepositoryTest(BaseRepositoryTest):
         self.assertIsNotNone(new_pool)
         self.assertEqual(0, len(new_pool.members))
 
+    def test_update_pool_members(self):
+        member1 = self.create_member(self.FAKE_UUID_1, self.FAKE_UUID_2,
+                                     self.pool.id, "10.0.0.1")
+        member2 = self.create_member(self.FAKE_UUID_3, self.FAKE_UUID_2,
+                                     self.pool.id, "10.0.0.2")
+        self.member_repo.update_pool_members(
+            self.session,
+            pool_id=self.pool.id,
+            operating_status=constants.OFFLINE)
+        new_member1 = self.member_repo.get(self.session, id=member1.id)
+        new_member2 = self.member_repo.get(self.session, id=member2.id)
+        self.assertEqual(constants.OFFLINE, new_member1.operating_status)
+        self.assertEqual(constants.OFFLINE, new_member2.operating_status)
+
 
 class SessionPersistenceRepositoryTest(BaseRepositoryTest):
 

octavia/tests/unit/controller/worker/flows/test_health_monitor_flows.py

@@ -36,6 +36,7 @@ class TestHealthMonitorFlows(base.TestCase):
 
         self.assertIn(constants.LISTENERS, health_mon_flow.requires)
         self.assertIn(constants.LOADBALANCER, health_mon_flow.requires)
+        self.assertIn(constants.POOL, health_mon_flow.requires)
 
         self.assertEqual(4, len(health_mon_flow.requires))
         self.assertEqual(0, len(health_mon_flow.provides))

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

@@ -2631,3 +2631,23 @@ class TestDatabaseTasks(base.TestCase):
             'TEST',
             id=POOL_ID,
             provisioning_status=constants.ERROR)
+
+    @mock.patch('octavia.db.repositories.MemberRepository.update_pool_members')
+    def test_update_pool_members_operating_status_in_db(
+            self,
+            mock_member_repo_update_pool_members,
+            mock_generate_uuid,
+            mock_LOG,
+            mock_get_session,
+            mock_loadbalancer_repo_update,
+            mock_listener_repo_update,
+            mock_amphora_repo_update,
+            mock_amphora_repo_delete):
+
+        update_members = database_tasks.UpdatePoolMembersOperatingStatusInDB()
+        update_members.execute(self.pool_mock, constants.ONLINE)
+
+        mock_member_repo_update_pool_members.assert_called_once_with(
+            'TEST',
+            POOL_ID,
+            operating_status=constants.ONLINE)