Add scatter gather utilities for cells

This adds utilities: scatter_gather_all_cells(),
scatter_gather_skip_cell0(), and scatter_gather_cells() for querying
cells in parallel using eventlet green threads.

Change-Id: I289932176e8029b0f9a76dbfb963f8ac218fdc06

1 files changed, 119 insertions, 0 deletions

diff --git a/nova/tests/unit/test_context.py b/nova/tests/unit/test_context.py
index 984b9c6c9f..387078557b 100644
--- a/nova/tests/unit/test_context.py
+++ b/nova/tests/unit/test_context.py
@@ -20,6 +20,7 @@ from nova import context
 from nova import exception
 from nova import objects
 from nova import test
+from nova.tests import fixtures as nova_fixtures
 from nova.tests import uuidsentinel as uuids
 
 
@@ -302,3 +303,121 @@ class ContextTestCase(test.NoDBTestCase):
             self.assertEqual(mock.sentinel.mq_conn_obj, cctxt.mq_connection)
         mock_create_cm.assert_not_called()
         mock_create_tport.assert_not_called()
+
+    @mock.patch('nova.context.target_cell')
+    @mock.patch('nova.objects.InstanceList.get_by_filters')
+    def test_scatter_gather_cells(self, mock_get_inst, mock_target_cell):
+        self.useFixture(nova_fixtures.SpawnIsSynchronousFixture())
+        ctxt = context.get_context()
+        mapping = objects.CellMapping(database_connection='fake://db',
+                                      transport_url='fake://mq',
+                                      uuid=uuids.cell)
+        mappings = objects.CellMappingList(objects=[mapping])
+
+        filters = {'deleted': False}
+        context.scatter_gather_cells(
+            ctxt, mappings, 60, objects.InstanceList.get_by_filters, filters,
+            sort_dir='foo')
+
+        mock_get_inst.assert_called_once_with(
+            mock_target_cell.return_value.__enter__.return_value, filters,
+            sort_dir='foo')
+
+    @mock.patch('nova.context.LOG.warning')
+    @mock.patch('eventlet.timeout.Timeout')
+    @mock.patch('eventlet.queue.LightQueue.get')
+    @mock.patch('nova.objects.InstanceList.get_by_filters')
+    def test_scatter_gather_cells_timeout(self, mock_get_inst,
+                                          mock_get_result, mock_timeout,
+                                          mock_log_warning):
+        # This is needed because we're mocking get_by_filters.
+        self.useFixture(nova_fixtures.SpawnIsSynchronousFixture())
+        ctxt = context.get_context()
+        mapping0 = objects.CellMapping(database_connection='fake://db0',
+                                       transport_url='none:///',
+                                       uuid=objects.CellMapping.CELL0_UUID)
+        mapping1 = objects.CellMapping(database_connection='fake://db1',
+                                       transport_url='fake://mq1',
+                                       uuid=uuids.cell1)
+        mappings = objects.CellMappingList(objects=[mapping0, mapping1])
+
+        # Simulate cell1 not responding.
+        mock_get_result.side_effect = [(mapping0.uuid,
+                                        mock.sentinel.instances),
+                                       exception.CellTimeout()]
+
+        results = context.scatter_gather_cells(
+            ctxt, mappings, 30, objects.InstanceList.get_by_filters)
+        self.assertEqual(2, len(results))
+        self.assertIn(mock.sentinel.instances, results.values())
+        self.assertIn(context.did_not_respond_sentinel, results.values())
+        mock_timeout.assert_called_once_with(30, exception.CellTimeout)
+        self.assertTrue(mock_log_warning.called)
+
+    @mock.patch('nova.context.LOG.exception')
+    @mock.patch('nova.objects.InstanceList.get_by_filters')
+    def test_scatter_gather_cells_exception(self, mock_get_inst,
+                                            mock_log_exception):
+        # This is needed because we're mocking get_by_filters.
+        self.useFixture(nova_fixtures.SpawnIsSynchronousFixture())
+        ctxt = context.get_context()
+        mapping0 = objects.CellMapping(database_connection='fake://db0',
+                                       transport_url='none:///',
+                                       uuid=objects.CellMapping.CELL0_UUID)
+        mapping1 = objects.CellMapping(database_connection='fake://db1',
+                                       transport_url='fake://mq1',
+                                       uuid=uuids.cell1)
+        mappings = objects.CellMappingList(objects=[mapping0, mapping1])
+
+        # Simulate cell1 raising an exception.
+        mock_get_inst.side_effect = [mock.sentinel.instances,
+                                     test.TestingException()]
+
+        results = context.scatter_gather_cells(
+            ctxt, mappings, 30, objects.InstanceList.get_by_filters)
+        self.assertEqual(2, len(results))
+        self.assertIn(mock.sentinel.instances, results.values())
+        self.assertIn(context.raised_exception_sentinel, results.values())
+        self.assertTrue(mock_log_exception.called)
+
+    @mock.patch('nova.context.scatter_gather_cells')
+    @mock.patch('nova.objects.CellMappingList.get_all')
+    def test_scatter_gather_all_cells(self, mock_get_all, mock_scatter):
+        ctxt = context.get_context()
+        mapping0 = objects.CellMapping(database_connection='fake://db0',
+                                       transport_url='none:///',
+                                       uuid=objects.CellMapping.CELL0_UUID)
+        mapping1 = objects.CellMapping(database_connection='fake://db1',
+                                       transport_url='fake://mq1',
+                                       uuid=uuids.cell1)
+        mock_get_all.return_value = objects.CellMappingList(
+            objects=[mapping0, mapping1])
+
+        filters = {'deleted': False}
+        context.scatter_gather_all_cells(
+            ctxt, objects.InstanceList.get_by_filters, filters, sort_dir='foo')
+
+        mock_scatter.assert_called_once_with(
+            ctxt, mock_get_all.return_value, 60,
+            objects.InstanceList.get_by_filters, filters, sort_dir='foo')
+
+    @mock.patch('nova.context.scatter_gather_cells')
+    @mock.patch('nova.objects.CellMappingList.get_all')
+    def test_scatter_gather_skip_cell0(self, mock_get_all, mock_scatter):
+        ctxt = context.get_context()
+        mapping0 = objects.CellMapping(database_connection='fake://db0',
+                                       transport_url='none:///',
+                                       uuid=objects.CellMapping.CELL0_UUID)
+        mapping1 = objects.CellMapping(database_connection='fake://db1',
+                                       transport_url='fake://mq1',
+                                       uuid=uuids.cell1)
+        mock_get_all.return_value = objects.CellMappingList(
+            objects=[mapping0, mapping1])
+
+        filters = {'deleted': False}
+        context.scatter_gather_skip_cell0(
+            ctxt, objects.InstanceList.get_by_filters, filters, sort_dir='foo')
+
+        mock_scatter.assert_called_once_with(
+            ctxt, [mapping1], 60, objects.InstanceList.get_by_filters, filters,
+            sort_dir='foo')