Merge "Move instance_get_*() to conductor"

7 files changed, 168 insertions, 38 deletions

diff --git a/nova/compute/manager.py b/nova/compute/manager.py
index 633c51843..498ec9829 100644
--- a/nova/compute/manager.py
+++ b/nova/compute/manager.py
@@ -417,6 +417,9 @@ class ComputeManager(manager.SchedulerDependentManager):
         """Initialization for a standalone compute service."""
         self.driver.init_host(host=self.host)
         context = nova.context.get_admin_context()
+
+        # NOTE(danms): this requires some care since conductor
+        # may not be up and fielding requests by the time compute is
         instances = self.db.instance_get_all_by_host(context, self.host)
 
         if CONF.defer_iptables_apply:
@@ -729,7 +732,8 @@ class ComputeManager(manager.SchedulerDependentManager):
             return
 
         filters = {'vm_state': vm_states.BUILDING}
-        building_insts = self.db.instance_get_all_by_filters(context, filters)
+        building_insts = self.conductor_api.instance_get_all_by_filters(
+            context, filters)
 
         for instance in building_insts:
             if timeutils.is_older_than(instance['created_at'], timeout):
@@ -2819,7 +2823,7 @@ class ComputeManager(manager.SchedulerDependentManager):
                     continue
             else:
                 # No more in our copy of uuids.  Pull from the DB.
-                db_instances = self.db.instance_get_all_by_host(
+                db_instances = self.conductor_api.instance_get_all_by_host(
                         context, self.host)
                 if not db_instances:
                     # None.. just return.
@@ -2842,7 +2846,7 @@ class ComputeManager(manager.SchedulerDependentManager):
     @manager.periodic_task
     def _poll_rebooting_instances(self, context):
         if CONF.reboot_timeout > 0:
-            instances = self.db.instance_get_all_hung_in_rebooting(
+            instances = self.conductor_api.instance_get_all_hung_in_rebooting(
                 context, CONF.reboot_timeout)
             self.driver.poll_rebooting_instances(CONF.reboot_timeout,
                                                  instances)
@@ -2927,11 +2931,9 @@ class ComputeManager(manager.SchedulerDependentManager):
         if CONF.instance_usage_audit:
             if not compute_utils.has_audit_been_run(context, self.host):
                 begin, end = utils.last_completed_audit_period()
-                instances = self.db.instance_get_active_by_window_joined(
-                                                            context,
-                                                            begin,
-                                                            end,
-                                                            host=self.host)
+                capi = self.conductor_api
+                instances = capi.instance_get_active_by_window_joined(
+                    context, begin, end, host=self.host)
                 num_instances = len(instances)
                 errors = 0
                 successes = 0
@@ -2978,7 +2980,8 @@ class ComputeManager(manager.SchedulerDependentManager):
             self._last_bw_usage_poll = curr_time
             LOG.info(_("Updating bandwidth usage cache"))
 
-            instances = self.db.instance_get_all_by_host(context, self.host)
+            instances = self.conductor_api.instance_get_all_by_host(context,
+                                                                    self.host)
             try:
                 bw_counters = self.driver.get_all_bw_counters(instances)
             except NotImplementedError:
@@ -3039,7 +3042,8 @@ class ComputeManager(manager.SchedulerDependentManager):
     def _get_host_volume_bdms(self, context, host):
         """Return all block device mappings on a compute host"""
         compute_host_bdms = []
-        instances = self.db.instance_get_all_by_host(context, self.host)
+        instances = self.conductor_api.instance_get_all_by_host(context,
+                                                                self.host)
         for instance in instances:
             instance_bdms = self._get_instance_volume_bdms(context, instance)
             compute_host_bdms.append(dict(instance=instance,
@@ -3128,7 +3132,8 @@ class ComputeManager(manager.SchedulerDependentManager):
         If the instance is not found on the hypervisor, but is in the database,
         then a stop() API will be called on the instance.
         """
-        db_instances = self.db.instance_get_all_by_host(context, self.host)
+        db_instances = self.conductor_api.instance_get_all_by_host(context,
+                                                                   self.host)
 
         num_vm_instances = self.driver.get_num_instances()
         num_db_instances = len(db_instances)
@@ -3258,7 +3263,8 @@ class ComputeManager(manager.SchedulerDependentManager):
             LOG.debug(_("CONF.reclaim_instance_interval <= 0, skipping..."))
             return
 
-        instances = self.db.instance_get_all_by_host(context, self.host)
+        instances = self.conductor_api.instance_get_all_by_host(context,
+                                                                self.host)
         for instance in instances:
             old_enough = (not instance['deleted_at'] or
                           timeutils.is_older_than(instance['deleted_at'],
@@ -3360,7 +3366,8 @@ class ComputeManager(manager.SchedulerDependentManager):
                 return True
             return False
         present_name_labels = set(self.driver.list_instances())
-        instances = self.db.instance_get_all_by_host(context, self.host)
+        instances = self.conductor_api.instance_get_all_by_host(context,
+                                                                self.host)
         return [i for i in instances if deleted_instance(i)]
 
     @contextlib.contextmanager
@@ -3421,7 +3428,7 @@ class ComputeManager(manager.SchedulerDependentManager):
         if CONF.image_cache_manager_interval == 0:
             return
 
-        all_instances = self.db.instance_get_all(context)
+        all_instances = self.conductor_api.instance_get_all(context)
 
         # Determine what other nodes use this storage
         storage_users.register_storage_use(CONF.instances_path, CONF.host)
diff --git a/nova/conductor/api.py b/nova/conductor/api.py
index 888bdfa9f..64e574b12 100644
--- a/nova/conductor/api.py
+++ b/nova/conductor/api.py
@@ -75,8 +75,28 @@ class LocalAPI(object):
     def instance_get_by_uuid(self, context, instance_uuid):
         return self._manager.instance_get_by_uuid(context, instance_uuid)
 
+    def instance_get_all(self, context):
+        return self.instance_get_all_by_filters(context, {})
+
     def instance_get_all_by_host(self, context, host):
-        return self._manager.instance_get_all_by_host(context, host)
+        return self.instance_get_all_by_filters(context, {'host': host})
+
+    def instance_get_all_by_filters(self, context, filters,
+                                    sort_key='created_at',
+                                    sort_dir='desc'):
+        return self._manager.instance_get_all_by_filters(context,
+                                                         filters,
+                                                         sort_key,
+                                                         sort_dir)
+
+    def instance_get_all_hung_in_rebooting(self, context, timeout):
+        return self._manager.instance_get_all_hung_in_rebooting(context,
+                                                                timeout)
+
+    def instance_get_active_by_window(self, context, begin, end=None,
+                                       project_id=None, host=None):
+        return self._manager.instance_get_active_by_window(
+            context, begin, end, project_id, host)
 
     def migration_get(self, context, migration_id):
         return self._manager.migration_get(context, migration_id)
@@ -185,8 +205,28 @@ class API(object):
         return self.conductor_rpcapi.instance_get_by_uuid(context,
                                                           instance_uuid)
 
+    def instance_get_all(self, context):
+        return self.instance_get_all_by_filters(context, {})
+
     def instance_get_all_by_host(self, context, host):
-        return self.conductor_rpcapi.instance_get_all_by_host(context, host)
+        return self.instance_get_all_by_filters(context, {'host': host})
+
+    def instance_get_all_by_filters(self, context, filters,
+                                    sort_key='created_at',
+                                    sort_dir='desc'):
+        return self.conductor_rpcapi.instance_get_all_by_filters(context,
+                                                                 filters,
+                                                                 sort_key,
+                                                                 sort_dir)
+
+    def instance_get_all_hung_in_rebooting(self, context, timeout):
+        return self.conductor_rpcapi.instance_get_all_hung_in_rebooting(
+            context, timeout)
+
+    def instance_get_active_by_window(self, context, begin, end=None,
+                                      project_id=None, host=None):
+        return self.conductor_rpcapi.instance_get_active_by_window(
+            context, begin, end, project_id, host)
 
     def migration_get(self, context, migration_id):
         return self.conductor_rpcapi.migration_get(context, migration_id)
diff --git a/nova/conductor/manager.py b/nova/conductor/manager.py
index 6b7344ed2..f5971b8d7 100644
--- a/nova/conductor/manager.py
+++ b/nova/conductor/manager.py
@@ -43,7 +43,7 @@ datetime_fields = ['launched_at', 'terminated_at']
 class ConductorManager(manager.SchedulerDependentManager):
     """Mission: TBD"""
 
-    RPC_API_VERSION = '1.14'
+    RPC_API_VERSION = '1.15'
 
     def __init__(self, *args, **kwargs):
         super(ConductorManager, self).__init__(service_name='conductor',
@@ -72,6 +72,7 @@ class ConductorManager(manager.SchedulerDependentManager):
         return jsonutils.to_primitive(
             self.db.instance_get_by_uuid(context, instance_uuid))
 
+    # NOTE(danms): This should go away in RPC version 2
     def instance_get_all_by_host(self, context, host):
         return jsonutils.to_primitive(
             self.db.instance_get_all_by_host(context.elevated(), host))
@@ -186,3 +187,21 @@ class ConductorManager(manager.SchedulerDependentManager):
             # NOTE(danms): This shouldn't happen
             raise exception.Invalid(_("Invalid block_device_mapping_destroy"
                                       " invocation"))
+
+    def instance_get_all_by_filters(self, context, filters, sort_key,
+                                    sort_dir):
+        result = self.db.instance_get_all_by_filters(context, filters,
+                                                     sort_key, sort_dir)
+        return jsonutils.to_primitive(result)
+
+    def instance_get_all_hung_in_rebooting(self, context, timeout):
+        result = self.db.instance_get_all_hung_in_rebooting(context, timeout)
+        return jsonutils.to_primitive(result)
+
+    def instance_get_active_by_window(self, context, begin, end=None,
+                                      project_id=None, host=None):
+        result = self.db.instance_get_active_by_window_joined(context,
+                                                              begin, end,
+                                                              project_id,
+                                                              host)
+        return jsonutils.to_primitive(result)
diff --git a/nova/conductor/rpcapi.py b/nova/conductor/rpcapi.py
index ceb38a425..92f9f9de7 100644
--- a/nova/conductor/rpcapi.py
+++ b/nova/conductor/rpcapi.py
@@ -43,6 +43,10 @@ class ConductorAPI(nova.openstack.common.rpc.proxy.RpcProxy):
     1.12 - Added block_device_mapping_update_or_create
     1.13 - Added block_device_mapping_get_all_by_instance
     1.14 - Added block_device_mapping_destroy
+    1.15 - Added instance_get_all_by_filters and
+           instance_get_all_hung_in_rebooting and
+           instance_get_active_by_window
+           Deprecated instance_get_all_by_host
     """
 
     BASE_RPC_API_VERSION = '1.0'
@@ -64,10 +68,6 @@ class ConductorAPI(nova.openstack.common.rpc.proxy.RpcProxy):
                             instance_uuid=instance_uuid)
         return self.call(context, msg, version='1.2')
 
-    def instance_get_all_by_host(self, context, host):
-        msg = self.make_msg('instance_get_all_by_host', host=host)
-        return self.call(context, msg, version='1.2')
-
     def migration_get(self, context, migration_id):
         msg = self.make_msg('migration_get', migration_id=migration_id)
         return self.call(context, msg, version='1.4')
@@ -171,3 +171,22 @@ class ConductorAPI(nova.openstack.common.rpc.proxy.RpcProxy):
                             instance=instance_p, volume_id=volume_id,
                             device_name=device_name)
         return self.call(context, msg, version='1.14')
+
+    def instance_get_all_by_filters(self, context, filters, sort_key,
+                                    sort_dir):
+        msg = self.make_msg('instance_get_all_by_filters',
+                            filters=filters, sort_key=sort_key,
+                            sort_dir=sort_dir)
+        return self.call(context, msg, version='1.15')
+
+    def instance_get_all_hung_in_rebooting(self, context, timeout):
+        msg = self.make_msg('instance_get_all_hung_in_rebooting',
+                            timeout=timeout)
+        return self.call(context, msg, version='1.15')
+
+    def instance_get_active_by_window(self, context, begin, end=None,
+                                      project_id=None, host=None):
+        msg = self.make_msg('instance_get_active_by_window',
+                            begin=begin, end=end, project_id=project_id,
+                            host=host)
+        return self.call(context, msg, version='1.15')
diff --git a/nova/tests/compute/test_compute.py b/nova/tests/compute/test_compute.py
index 104fd4e68..57c234734 100644
--- a/nova/tests/compute/test_compute.py
+++ b/nova/tests/compute/test_compute.py
@@ -2732,10 +2732,10 @@ class ComputeTestCase(BaseTestCase):
         self.flags(running_deleted_instance_timeout=3600,
                    running_deleted_instance_action='reap')
 
-        self.mox.StubOutWithMock(self.compute.db, "instance_get_all_by_host")
-        self.compute.db.instance_get_all_by_host(admin_context,
-                                                 self.compute.host
-                                                ).AndReturn([instance])
+        self.mox.StubOutWithMock(self.compute.conductor_api,
+                                 "instance_get_all_by_host")
+        self.compute.conductor_api.instance_get_all_by_host(
+            admin_context, self.compute.host).AndReturn([instance])
 
         bdms = []
 
@@ -2771,9 +2771,10 @@ class ComputeTestCase(BaseTestCase):
         timeutils.is_older_than('sometimeago',
                     CONF.running_deleted_instance_timeout).AndReturn(True)
 
-        self.mox.StubOutWithMock(self.compute.db, "instance_get_all_by_host")
-        self.compute.db.instance_get_all_by_host('context',
-                                                 'host').AndReturn(
+        self.mox.StubOutWithMock(self.compute.conductor_api,
+                                 "instance_get_all_by_host")
+        self.compute.conductor_api.instance_get_all_by_host('context',
+                                                            'host').AndReturn(
                                                                 [instance1,
                                                                  instance2])
         self.mox.ReplayAll()
@@ -2814,7 +2815,7 @@ class ComputeTestCase(BaseTestCase):
             self.assertEqual(instance, call_info['expected_instance'])
             call_info['get_nw_info'] += 1
 
-        self.stubs.Set(db, 'instance_get_all_by_host',
+        self.stubs.Set(self.compute.conductor_api, 'instance_get_all_by_host',
                 fake_instance_get_all_by_host)
         self.stubs.Set(db, 'instance_get_by_uuid',
                 fake_instance_get_by_uuid)
diff --git a/nova/tests/conductor/test_conductor.py b/nova/tests/conductor/test_conductor.py
index 23930770a..a7caa3ac9 100644
--- a/nova/tests/conductor/test_conductor.py
+++ b/nova/tests/conductor/test_conductor.py
@@ -116,13 +116,6 @@ class _BaseTestCase(object):
         self.assertEqual(orig_instance['name'],
                          copy_instance['name'])
 
-    def test_instance_get_all_by_host(self):
-        orig_instance = jsonutils.to_primitive(self._create_fake_instance())
-        all_instances = self.conductor.instance_get_all_by_host(
-            self.context, orig_instance['host'])
-        self.assertEqual(orig_instance['name'],
-                         all_instances[0]['name'])
-
     def _setup_aggregate_with_host(self):
         aggregate_ref = db.aggregate_create(self.context.elevated(),
                 {'name': 'foo', 'availability_zone': 'foo'})
@@ -279,6 +272,22 @@ class _BaseTestCase(object):
             self.context, fake_inst)
         self.assertEqual(result, 'fake-result')
 
+    def test_instance_get_all_hung_in_rebooting(self):
+        self.mox.StubOutWithMock(db, 'instance_get_all_hung_in_rebooting')
+        db.instance_get_all_hung_in_rebooting(self.context, 123)
+        self.mox.ReplayAll()
+        self.conductor.instance_get_all_hung_in_rebooting(self.context, 123)
+
+    def test_instance_get_active_by_window(self):
+        self.mox.StubOutWithMock(db, 'instance_get_active_by_window_joined')
+        db.instance_get_active_by_window_joined(self.context, 'fake-begin',
+                                                'fake-end', 'fake-proj',
+                                                'fake-host')
+        self.mox.ReplayAll()
+        self.conductor.instance_get_active_by_window(self.context,
+                                                     'fake-begin', 'fake-end',
+                                                     'fake-proj', 'fake-host')
+
 
 class ConductorTestCase(_BaseTestCase, test.TestCase):
     """Conductor Manager Tests"""
@@ -333,6 +342,15 @@ class ConductorTestCase(_BaseTestCase, test.TestCase):
                                                     instance=fake_inst,
                                                     volume_id='fake-volume')
 
+    def test_instance_get_all_by_filters(self):
+        filters = {'foo': 'bar'}
+        self.mox.StubOutWithMock(db, 'instance_get_all_by_filters')
+        db.instance_get_all_by_filters(self.context, filters,
+                                       'fake-key', 'fake-sort')
+        self.mox.ReplayAll()
+        self.conductor.instance_get_all_by_filters(self.context, filters,
+                                                   'fake-key', 'fake-sort')
+
 
 class ConductorRPCAPITestCase(_BaseTestCase, test.TestCase):
     """Conductor RPC API Tests"""
@@ -385,6 +403,15 @@ class ConductorRPCAPITestCase(_BaseTestCase, test.TestCase):
                                                     instance=fake_inst,
                                                     volume_id='fake-volume')
 
+    def test_instance_get_all_by_filters(self):
+        filters = {'foo': 'bar'}
+        self.mox.StubOutWithMock(db, 'instance_get_all_by_filters')
+        db.instance_get_all_by_filters(self.context, filters,
+                                       'fake-key', 'fake-sort')
+        self.mox.ReplayAll()
+        self.conductor.instance_get_all_by_filters(self.context, filters,
+                                                   'fake-key', 'fake-sort')
+
 
 class ConductorAPITestCase(_BaseTestCase, test.TestCase):
     """Conductor API Tests"""
@@ -450,6 +477,20 @@ class ConductorAPITestCase(_BaseTestCase, test.TestCase):
         self.conductor.block_device_mapping_destroy_by_instance_and_volume(
             self.context, fake_inst, 'fake-volume')
 
+    def test_instance_get_all(self):
+        self.mox.StubOutWithMock(db, 'instance_get_all_by_filters')
+        db.instance_get_all_by_filters(self.context, {}, 'created_at', 'desc')
+        db.instance_get_all_by_filters(self.context, {'host': 'fake-host'},
+                                       'created_at', 'desc')
+        db.instance_get_all_by_filters(self.context, {'name': 'fake-inst'},
+                                       'updated_at', 'asc')
+        self.mox.ReplayAll()
+        self.conductor.instance_get_all(self.context)
+        self.conductor.instance_get_all_by_host(self.context, 'fake-host')
+        self.conductor.instance_get_all_by_filters(self.context,
+                                                   {'name': 'fake-inst'},
+                                                   'updated_at', 'asc')
+
 
 class ConductorLocalAPITestCase(ConductorAPITestCase):
     """Conductor LocalAPI Tests"""
diff --git a/nova/tests/test_imagecache.py b/nova/tests/test_imagecache.py
index ebda2c718..affab4e29 100644
--- a/nova/tests/test_imagecache.py
+++ b/nova/tests/test_imagecache.py
@@ -27,6 +27,7 @@ import time
 from nova import test
 
 from nova.compute import vm_states
+from nova import conductor
 from nova import db
 from nova.openstack.common import cfg
 from nova.openstack.common import importutils
@@ -928,7 +929,7 @@ class ImageCacheManagerTestCase(test.TestCase):
     def test_compute_manager(self):
         was = {'called': False}
 
-        def fake_get_all(context):
+        def fake_get_all(context, *args, **kwargs):
             was['called'] = True
             return [{'image_ref': '1',
                      'host': CONF.host,
@@ -946,7 +947,9 @@ class ImageCacheManagerTestCase(test.TestCase):
         with utils.tempdir() as tmpdir:
             self.flags(instances_path=tmpdir)
 
-            self.stubs.Set(db, 'instance_get_all', fake_get_all)
+            self.stubs.Set(db, 'instance_get_all_by_filters', fake_get_all)
             compute = importutils.import_object(CONF.compute_manager)
+            self.flags(use_local=True, group='conductor')
+            compute.conductor_api = conductor.API()
             compute._run_image_cache_manager_pass(None)
             self.assertTrue(was['called'])