Use Instance Objects for Start/Stop

This patch makes the start and stop operations use the Instance
object instead of passing SQLA-derived dicts over RPC. Until
something more sophisticated is needed (and developed), it also
adds a nova.object.register_all() function which just triggers
imports of all the object models so that they are registered.
When adding a new object type, that register function should be
updated appropriately.

Related to bp/unified-object-model

Change-Id: I3c8d9cba07d34097a279502062906de802d19d1f

11 files changed, 258 insertions, 124 deletions

diff --git a/nova/api/ec2/cloud.py b/nova/api/ec2/cloud.py
index da0a52caa..af50868d7 100644
--- a/nova/api/ec2/cloud.py
+++ b/nova/api/ec2/cloud.py
@@ -43,6 +43,7 @@ from nova import exception
 from nova.image import s3
 from nova import network
 from nova.network.security_group import quantum_driver
+from nova.objects import instance as instance_obj
 from nova.openstack.common import log as logging
 from nova.openstack.common import timeutils
 from nova import quota
@@ -206,6 +207,16 @@ def _format_mappings(properties, result):
         result['blockDeviceMapping'] = mappings
 
 
+def db_to_inst_obj(context, db_instance):
+    # NOTE(danms): This is a temporary helper method for converting
+    # Instance DB objects to NovaObjects without needing to re-query.
+    inst_obj = instance_obj.Instance._from_db_object(
+        instance_obj.Instance(), db_instance,
+        expected_attrs=['system_metadata', 'metadata'])
+    inst_obj._context = context
+    return inst_obj
+
+
 class CloudController(object):
     """CloudController provides the critical dispatch between
  inbound API calls through the endpoint and messages
@@ -1340,13 +1351,18 @@ class CloudController(object):
             block_device_mapping=kwargs.get('block_device_mapping', {}))
         return self._format_run_instances(context, resv_id)
 
-    def _ec2_ids_to_instances(self, context, instance_id):
+    def _ec2_ids_to_instances(self, context, instance_id, objects=False):
         """Get all instances first, to prevent partial executions."""
         instances = []
+        extra = ['system_metadata', 'metadata']
         for ec2_id in instance_id:
             validate_ec2_id(ec2_id)
             instance_uuid = ec2utils.ec2_inst_id_to_uuid(context, ec2_id)
-            instance = self.compute_api.get(context, instance_uuid)
+            if objects:
+                instance = instance_obj.Instance.get_by_uuid(
+                    context, instance_uuid, expected_attrs=extra)
+            else:
+                instance = self.compute_api.get(context, instance_uuid)
             instances.append(instance)
         return instances
 
@@ -1372,7 +1388,7 @@ class CloudController(object):
     def stop_instances(self, context, instance_id, **kwargs):
         """Stop each instances in instance_id.
         Here instance_id is a list of instance ids"""
-        instances = self._ec2_ids_to_instances(context, instance_id)
+        instances = self._ec2_ids_to_instances(context, instance_id, True)
         LOG.debug(_("Going to stop instances"))
         for instance in instances:
             self.compute_api.stop(context, instance)
@@ -1381,7 +1397,7 @@ class CloudController(object):
     def start_instances(self, context, instance_id, **kwargs):
         """Start each instances in instance_id.
         Here instance_id is a list of instance ids"""
-        instances = self._ec2_ids_to_instances(context, instance_id)
+        instances = self._ec2_ids_to_instances(context, instance_id, True)
         LOG.debug(_("Going to start instances"))
         for instance in instances:
             self.compute_api.start(context, instance)
@@ -1635,7 +1651,8 @@ class CloudController(object):
 
             if vm_state == vm_states.ACTIVE:
                 restart_instance = True
-                self.compute_api.stop(context, instance)
+                inst_obj = db_to_inst_obj(context, instance)
+                self.compute_api.stop(context, inst_obj)
 
             # wait instance for really stopped
             start_time = time.time()
@@ -1677,7 +1694,8 @@ class CloudController(object):
         ec2_id = ec2utils.glance_id_to_ec2_id(context, new_image['id'])
 
         if restart_instance:
-            self.compute_api.start(context, instance)
+            inst_obj = db_to_inst_obj(context, instance)
+            self.compute_api.start(context, inst_obj)
 
         return {'imageId': ec2_id}
 
diff --git a/nova/api/openstack/compute/contrib/server_start_stop.py b/nova/api/openstack/compute/contrib/server_start_stop.py
index c4d0d5c9e..2803cd04b 100644
--- a/nova/api/openstack/compute/contrib/server_start_stop.py
+++ b/nova/api/openstack/compute/contrib/server_start_stop.py
@@ -20,6 +20,7 @@ from nova.api.openstack import extensions
 from nova.api.openstack import wsgi
 from nova import compute
 from nova import exception
+from nova.objects import instance as instance_obj
 from nova.openstack.common import log as logging
 
 
@@ -33,7 +34,9 @@ class ServerStartStopActionController(wsgi.Controller):
 
     def _get_instance(self, context, instance_uuid):
         try:
-            return self.compute_api.get(context, instance_uuid)
+            attrs = ['system_metadata', 'metadata']
+            return instance_obj.Instance.get_by_uuid(context, instance_uuid,
+                                                     expected_attrs=attrs)
         except exception.NotFound:
             msg = _("Instance not found")
             raise webob.exc.HTTPNotFound(explanation=msg)
diff --git a/nova/cmd/compute.py b/nova/cmd/compute.py
index 0aae286a4..89b7e705a 100644
--- a/nova/cmd/compute.py
+++ b/nova/cmd/compute.py
@@ -23,9 +23,12 @@ import traceback
 
 from oslo.config import cfg
 
+from nova.conductor import rpcapi as conductor_rpcapi
 from nova import config
 import nova.db.api
 from nova import exception
+from nova import objects
+from nova.objects import base as objects_base
 from nova.openstack.common import log as logging
 from nova import service
 from nova import utils
@@ -50,12 +53,15 @@ def block_db_access():
 
 
 def main():
+    objects.register_all()
     config.parse_args(sys.argv)
     logging.setup('nova')
     utils.monkey_patch()
 
     if not CONF.conductor.use_local:
         block_db_access()
+        objects_base.NovaObject.indirection_api = \
+            conductor_rpcapi.ConductorAPI()
 
     server = service.Service.create(binary='nova-compute',
                                     topic=CONF.compute_topic,
diff --git a/nova/cmd/conductor.py b/nova/cmd/conductor.py
index b9723f2d2..3fc25eb8c 100644
--- a/nova/cmd/conductor.py
+++ b/nova/cmd/conductor.py
@@ -21,6 +21,7 @@ import sys
 from oslo.config import cfg
 
 from nova import config
+from nova import objects
 from nova.openstack.common import log as logging
 from nova import service
 from nova import utils
@@ -30,6 +31,7 @@ CONF.import_opt('topic', 'nova.conductor.api', group='conductor')
 
 
 def main():
+    objects.register_all()
     config.parse_args(sys.argv)
     logging.setup("nova")
     utils.monkey_patch()
diff --git a/nova/compute/api.py b/nova/compute/api.py
index f676c9797..9ef81ef2b 100644
--- a/nova/compute/api.py
+++ b/nova/compute/api.py
@@ -49,6 +49,7 @@ from nova import network
 from nova.network.security_group import openstack_driver
 from nova.network.security_group import security_group_base
 from nova import notifications
+from nova.objects import instance as instance_obj
 from nova.openstack.common import excutils
 from nova.openstack.common import jsonutils
 from nova.openstack.common import log as logging
@@ -1343,10 +1344,16 @@ class API(base.Base):
         """Stop an instance."""
         LOG.debug(_("Going to try to stop instance"), instance=instance)
 
-        instance = self.update(context, instance,
-                    task_state=task_states.POWERING_OFF,
-                    expected_task_state=None,
-                    progress=0)
+        # NOTE(danms): Temporary transition to objects. Remove after
+        # compute/manager.py _sync_instance_power_state() is migrated.
+        if isinstance(instance, dict):
+            instance = instance_obj.Instance._from_db_object(
+                instance_obj.Instance(), instance)
+            instance._context = context
+
+        instance.task_state = task_states.POWERING_OFF
+        instance.progress = 0
+        instance.save(expected_task_state=None)
 
         self._record_action_start(context, instance, instance_actions.STOP)
 
@@ -1360,9 +1367,8 @@ class API(base.Base):
         """Start an instance."""
         LOG.debug(_("Going to try to start instance"), instance=instance)
 
-        instance = self.update(context, instance,
-                               task_state=task_states.POWERING_ON,
-                               expected_task_state=None)
+        instance.task_state = task_states.POWERING_ON
+        instance.save(expected_task_state=None)
 
         self._record_action_start(context, instance, instance_actions.START)
         # TODO(yamahata): injected_files isn't supported right now.
diff --git a/nova/compute/manager.py b/nova/compute/manager.py
index 2390eb3c3..8c593bda5 100755
--- a/nova/compute/manager.py
+++ b/nova/compute/manager.py
@@ -60,6 +60,7 @@ from nova import manager
 from nova import network
 from nova.network import model as network_model
 from nova.network.security_group import openstack_driver
+from nova.objects import instance as instance_obj
 from nova.openstack.common import excutils
 from nova.openstack.common import jsonutils
 from nova.openstack.common import log as logging
@@ -266,6 +267,27 @@ def wrap_instance_event(function):
     return decorated_function
 
 
+# TODO(danms): Remove me after havana
+def object_compat(function):
+    """Wraps a method that expects a new-world instance
+
+    This provides compatibility for callers passing old-style dict
+    instances.
+    """
+
+    @functools.wraps(function)
+    def decorated_function(self, context, **kwargs):
+        metas = ['metadata', 'system_metadata']
+        instance = kwargs['instance']
+        if isinstance(instance, dict):
+            kwargs['instance'] = instance_obj.Instance._from_db_object(
+                instance_obj.Instance(), instance, expected_attrs=metas)
+            kwargs['instance']._context = context
+        return function(self, context, **kwargs)
+
+    return decorated_function
+
+
 def _get_image_meta(context, image_ref):
     image_service, image_id = glance.get_remote_image_service(context,
                                                               image_ref)
@@ -328,7 +350,7 @@ class ComputeVirtAPI(virtapi.VirtAPI):
 class ComputeManager(manager.SchedulerDependentManager):
     """Manages the running instances from creation to destruction."""
 
-    RPC_API_VERSION = '2.28'
+    RPC_API_VERSION = '2.29'
 
     def __init__(self, compute_driver=None, *args, **kwargs):
         """Load configuration options and connect to the hypervisor."""
@@ -1457,6 +1479,7 @@ class ComputeManager(manager.SchedulerDependentManager):
     # NOTE(johannes): This is probably better named power_off_instance
     # so it matches the driver method, but because of other issues, we
     # can't use that name in grizzly.
+    @object_compat
     @exception.wrap_exception(notifier=notifier, publisher_id=publisher_id())
     @reverts_task_state
     @wrap_instance_event
@@ -1466,17 +1489,17 @@ class ComputeManager(manager.SchedulerDependentManager):
         self._notify_about_instance_usage(context, instance, "power_off.start")
         self.driver.power_off(instance)
         current_power_state = self._get_power_state(context, instance)
-        instance = self._instance_update(context, instance['uuid'],
-                power_state=current_power_state,
-                vm_state=vm_states.STOPPED,
-                expected_task_state=(task_states.POWERING_OFF,
-                                     task_states.STOPPING),
-                task_state=None)
+        instance.power_state = current_power_state
+        instance.vm_state = vm_states.STOPPED
+        instance.task_state = None
+        instance.save(expected_task_state=(task_states.POWERING_OFF,
+                                           task_states.STOPPING))
         self._notify_about_instance_usage(context, instance, "power_off.end")
 
     # NOTE(johannes): This is probably better named power_on_instance
     # so it matches the driver method, but because of other issues, we
     # can't use that name in grizzly.
+    @object_compat
     @exception.wrap_exception(notifier=notifier, publisher_id=publisher_id())
     @reverts_task_state
     @wrap_instance_event
@@ -1486,12 +1509,11 @@ class ComputeManager(manager.SchedulerDependentManager):
         self._notify_about_instance_usage(context, instance, "power_on.start")
         self.driver.power_on(instance)
         current_power_state = self._get_power_state(context, instance)
-        instance = self._instance_update(context, instance['uuid'],
-                power_state=current_power_state,
-                vm_state=vm_states.ACTIVE,
-                task_state=None,
-                expected_task_state=(task_states.POWERING_ON,
-                                     task_states.STARTING))
+        instance.power_state = current_power_state
+        instance.vm_state = vm_states.ACTIVE
+        instance.task_state = None
+        instance.save(expected_task_state=(task_states.POWERING_ON,
+                                           task_states.STARTING))
         self._notify_about_instance_usage(context, instance, "power_on.end")
 
     @exception.wrap_exception(notifier=notifier, publisher_id=publisher_id())
@@ -1709,7 +1731,7 @@ class ComputeManager(manager.SchedulerDependentManager):
                                  task_state=task_states.STOPPING,
                                  terminated_at=timeutils.utcnow(),
                                  progress=0)
-                self.stop_instance(context, instance)
+                self.stop_instance(context, instance=instance)
 
             self._notify_about_instance_usage(
                     context, instance, "rebuild.end",
@@ -2320,7 +2342,7 @@ class ComputeManager(manager.SchedulerDependentManager):
                 instance = self._instance_update(
                         context, instance['uuid'],
                         task_state=task_states.STOPPING)
-                self.stop_instance(context, instance)
+                self.stop_instance(context, instance=instance)
 
             self._notify_about_instance_usage(
                     context, instance, "resize.revert.end")
diff --git a/nova/compute/rpcapi.py b/nova/compute/rpcapi.py
index a8d7eaa47..ac3003cce 100644
--- a/nova/compute/rpcapi.py
+++ b/nova/compute/rpcapi.py
@@ -21,8 +21,10 @@ Client side of the compute RPC API.
 from oslo.config import cfg
 
 from nova import exception
+from nova.objects import base as objects_base
 from nova.openstack.common import jsonutils
 from nova.openstack.common import rpc
+import nova.openstack.common.rpc
 import nova.openstack.common.rpc.proxy
 
 rpcapi_opts = [
@@ -166,6 +168,8 @@ class ComputeAPI(nova.openstack.common.rpc.proxy.RpcProxy):
         2.27 - Adds 'reservations' to terminate_instance() and
                soft_delete_instance()
         2.28 - Adds check_instance_shared_storage()
+        2.29 - Made start_instance() and stop_instance() take new-world
+               instance objects
     '''
 
     #
@@ -181,7 +185,8 @@ class ComputeAPI(nova.openstack.common.rpc.proxy.RpcProxy):
     def __init__(self):
         super(ComputeAPI, self).__init__(
                 topic=CONF.compute_topic,
-                default_version=self.BASE_RPC_API_VERSION)
+                default_version=self.BASE_RPC_API_VERSION,
+                serializer=objects_base.NovaObjectSerializer())
 
     def add_aggregate_host(self, ctxt, aggregate, host_param, host,
                            slave_info=None):
@@ -581,17 +586,17 @@ class ComputeAPI(nova.openstack.common.rpc.proxy.RpcProxy):
                 topic=_compute_topic(self.topic, ctxt, None, instance))
 
     def start_instance(self, ctxt, instance):
-        instance_p = jsonutils.to_primitive(instance)
         self.cast(ctxt, self.make_msg('start_instance',
-                instance=instance_p),
-                topic=_compute_topic(self.topic, ctxt, None, instance))
+                instance=instance),
+                topic=_compute_topic(self.topic, ctxt, None, instance),
+                version='2.29')
 
     def stop_instance(self, ctxt, instance, cast=True):
         rpc_method = self.cast if cast else self.call
-        instance_p = jsonutils.to_primitive(instance)
         return rpc_method(ctxt, self.make_msg('stop_instance',
-                instance=instance_p),
-                topic=_compute_topic(self.topic, ctxt, None, instance))
+                instance=instance),
+                topic=_compute_topic(self.topic, ctxt, None, instance),
+                version='2.29')
 
     def suspend_instance(self, ctxt, instance):
         instance_p = jsonutils.to_primitive(instance)
diff --git a/nova/objects/__init__.py b/nova/objects/__init__.py
index 67f4db51a..e39f0154c 100644
--- a/nova/objects/__init__.py
+++ b/nova/objects/__init__.py
@@ -11,3 +11,10 @@
 #    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 #    License for the specific language governing permissions and limitations
 #    under the License.
+
+
+def register_all():
+    # NOTE(danms): You must make sure your object gets imported in this
+    # function in order for it to be registered by services that may
+    # need to receive it via RPC.
+    __import__('nova.objects.instance')
diff --git a/nova/tests/api/openstack/compute/contrib/test_server_start_stop.py b/nova/tests/api/openstack/compute/contrib/test_server_start_stop.py
index eb708a574..31b832084 100644
--- a/nova/tests/api/openstack/compute/contrib/test_server_start_stop.py
+++ b/nova/tests/api/openstack/compute/contrib/test_server_start_stop.py
@@ -17,13 +17,19 @@ import webob
 
 from nova.api.openstack.compute.contrib import server_start_stop
 from nova.compute import api as compute_api
+from nova import db
 from nova import exception
 from nova import test
 from nova.tests.api.openstack import fakes
 
 
-def fake_compute_api_get(self, context, instance_id):
-    return {'id': 1, 'uuid': instance_id}
+def fake_instance_get(self, context, instance_id):
+    result = fakes.stub_instance(id=1, uuid=instance_id)
+    result['created_at'] = None
+    result['deleted_at'] = None
+    result['updated_at'] = None
+    result['deleted'] = 0
+    return result
 
 
 def fake_start_stop_not_ready(self, context, instance):
@@ -37,7 +43,7 @@ class ServerStartStopTest(test.TestCase):
         self.controller = server_start_stop.ServerStartStopActionController()
 
     def test_start(self):
-        self.stubs.Set(compute_api.API, 'get', fake_compute_api_get)
+        self.stubs.Set(db, 'instance_get_by_uuid', fake_instance_get)
         self.mox.StubOutWithMock(compute_api.API, 'start')
         compute_api.API.start(mox.IgnoreArg(), mox.IgnoreArg())
         self.mox.ReplayAll()
@@ -47,7 +53,7 @@ class ServerStartStopTest(test.TestCase):
         self.controller._start_server(req, 'test_inst', body)
 
     def test_start_not_ready(self):
-        self.stubs.Set(compute_api.API, 'get', fake_compute_api_get)
+        self.stubs.Set(db, 'instance_get_by_uuid', fake_instance_get)
         self.stubs.Set(compute_api.API, 'start', fake_start_stop_not_ready)
         req = fakes.HTTPRequest.blank('/v2/fake/servers/test_inst/action')
         body = dict(start="")
@@ -55,7 +61,7 @@ class ServerStartStopTest(test.TestCase):
             self.controller._start_server, req, 'test_inst', body)
 
     def test_stop(self):
-        self.stubs.Set(compute_api.API, 'get', fake_compute_api_get)
+        self.stubs.Set(db, 'instance_get_by_uuid', fake_instance_get)
         self.mox.StubOutWithMock(compute_api.API, 'stop')
         compute_api.API.stop(mox.IgnoreArg(), mox.IgnoreArg())
         self.mox.ReplayAll()
@@ -65,7 +71,7 @@ class ServerStartStopTest(test.TestCase):
         self.controller._stop_server(req, 'test_inst', body)
 
     def test_stop_not_ready(self):
-        self.stubs.Set(compute_api.API, 'get', fake_compute_api_get)
+        self.stubs.Set(db, 'instance_get_by_uuid', fake_instance_get)
         self.stubs.Set(compute_api.API, 'stop', fake_start_stop_not_ready)
         req = fakes.HTTPRequest.blank('/v2/fake/servers/test_inst/action')
         body = dict(start="")
diff --git a/nova/tests/compute/test_compute.py b/nova/tests/compute/test_compute.py
index fc3e0126b..211838ea1 100644
--- a/nova/tests/compute/test_compute.py
+++ b/nova/tests/compute/test_compute.py
@@ -34,6 +34,7 @@ import nova
 from nova import compute
 from nova.compute import api as compute_api
 from nova.compute import flavors
+from nova.compute import instance_actions
 from nova.compute import manager as compute_manager
 from nova.compute import power_state
 from nova.compute import rpcapi as compute_rpcapi
@@ -48,6 +49,7 @@ from nova.image import glance
 from nova.network import api as network_api
 from nova.network import model as network_model
 from nova.network.security_group import openstack_driver
+from nova.objects import instance as instance_obj
 from nova.openstack.common import importutils
 from nova.openstack.common import jsonutils
 from nova.openstack.common import log as logging
@@ -62,6 +64,7 @@ from nova import quota
 from nova import test
 from nova.tests.compute import fake_resource_tracker
 from nova.tests.db import fakes as db_fakes
+from nova.tests import fake_instance
 from nova.tests import fake_instance_actions
 from nova.tests import fake_network
 from nova.tests import fake_network_cache_model
@@ -236,6 +239,7 @@ class BaseTestCase(test.TestCase):
 
         inst = {}
         inst['vm_state'] = vm_states.ACTIVE
+        inst['task_state'] = None
         inst['image_ref'] = FAKE_IMAGE_REF
         inst['reservation_id'] = 'r-fakeres'
         inst['user_id'] = self.user_id
@@ -252,11 +256,27 @@ class BaseTestCase(test.TestCase):
         inst['architecture'] = 'x86_64'
         inst['os_type'] = 'Linux'
         inst['system_metadata'] = make_fake_sys_meta()
+        inst['locked'] = False
         inst.update(params)
         _create_service_entries(self.context.elevated(),
                 {'fake_zone': [inst['host']]})
         return db.instance_create(self.context, inst)
 
+    def _create_instance_obj(self, params=None, type_name='m1.tiny'):
+        """Create a test instance object."""
+        instance = instance_obj.Instance()
+        instance.uuid = uuidutils.generate_uuid()
+        instance.cell_name = 'api!child'
+
+        def _fake_db_create(_ctxt, inst):
+            for k, v in inst.items():
+                setattr(instance, k, v)
+            return instance
+
+        self.stubs.Set(db, 'instance_create', _fake_db_create)
+        return self._create_fake_instance(params=params,
+                type_name=type_name)
+
     def _create_instance(self, params=None, type_name='m1.tiny'):
         """Create a test instance. Returns uuid."""
         return self._create_fake_instance(params, type_name=type_name)
@@ -691,6 +711,16 @@ class ComputeTestCase(BaseTestCase):
         self.assertTrue(called['finished'])
         self.assertEqual('An unknown exception occurred.', called['message'])
 
+    def test_object_compat(self):
+        db_inst = fake_instance.fake_db_instance()
+
+        @compute_manager.object_compat
+        def test_fn(_self, context, instance):
+            self.assertTrue(isinstance(instance, instance_obj.Instance))
+            self.assertEqual(instance.uuid, db_inst['uuid'])
+
+        test_fn(None, self.context, instance=db_inst)
+
     def test_create_instance_with_img_ref_associates_config_drive(self):
         # Make sure create associates a config drive.
 
@@ -1238,7 +1268,12 @@ class ComputeTestCase(BaseTestCase):
         self.compute.run_instance(self.context, instance=instance)
         db.instance_update(self.context, instance['uuid'],
                            {"task_state": task_states.POWERING_OFF})
-        self.compute.stop_instance(self.context, instance=instance)
+        inst_uuid = instance['uuid']
+        extra = ['system_metadata', 'metadata']
+        inst_obj = instance_obj.Instance.get_by_uuid(self.context,
+                                                     inst_uuid,
+                                                     expected_attrs=extra)
+        self.compute.stop_instance(self.context, instance=inst_obj)
         self.compute.terminate_instance(self.context, instance=instance)
 
     def test_start(self):
@@ -1247,10 +1282,15 @@ class ComputeTestCase(BaseTestCase):
         self.compute.run_instance(self.context, instance=instance)
         db.instance_update(self.context, instance['uuid'],
                            {"task_state": task_states.POWERING_OFF})
-        self.compute.stop_instance(self.context, instance=instance)
-        db.instance_update(self.context, instance['uuid'],
-                           {"task_state": task_states.POWERING_ON})
-        self.compute.start_instance(self.context, instance=instance)
+        extra = ['system_metadata', 'metadata']
+        inst_uuid = instance['uuid']
+        inst_obj = instance_obj.Instance.get_by_uuid(self.context,
+                                                     inst_uuid,
+                                                     expected_attrs=extra)
+        self.compute.stop_instance(self.context, instance=inst_obj)
+        inst_obj.task_state = task_states.POWERING_ON
+        inst_obj.save(self.context)
+        self.compute.start_instance(self.context, instance=inst_obj)
         self.compute.terminate_instance(self.context, instance=instance)
 
     def test_stop_start_no_image(self):
@@ -1259,10 +1299,15 @@ class ComputeTestCase(BaseTestCase):
         self.compute.run_instance(self.context, instance=instance)
         db.instance_update(self.context, instance['uuid'],
                            {"task_state": task_states.POWERING_OFF})
-        self.compute.stop_instance(self.context, instance=instance)
-        db.instance_update(self.context, instance['uuid'],
-                           {"task_state": task_states.POWERING_ON})
-        self.compute.start_instance(self.context, instance=instance)
+        extra = ['system_metadata', 'metadata']
+        inst_uuid = instance['uuid']
+        inst_obj = instance_obj.Instance.get_by_uuid(self.context,
+                                                     inst_uuid,
+                                                     expected_attrs=extra)
+        self.compute.stop_instance(self.context, instance=inst_obj)
+        inst_obj.task_state = task_states.POWERING_ON
+        inst_obj.save(self.context)
+        self.compute.start_instance(self.context, instance=inst_obj)
         self.compute.terminate_instance(self.context, instance=instance)
 
     def test_rescue(self):
@@ -1338,11 +1383,15 @@ class ComputeTestCase(BaseTestCase):
 
         instance = jsonutils.to_primitive(self._create_fake_instance())
         self.compute.run_instance(self.context, instance=instance)
-        db.instance_update(self.context, instance['uuid'],
-                           {"task_state": task_states.POWERING_ON})
-        self.compute.start_instance(self.context, instance=instance)
+        extra = ['system_metadata', 'metadata']
+        inst_obj = instance_obj.Instance.get_by_uuid(self.context,
+                                                     instance['uuid'],
+                                                     expected_attrs=extra)
+        inst_obj.task_state = task_states.POWERING_ON
+        inst_obj.save(self.context)
+        self.compute.start_instance(self.context, instance=inst_obj)
         self.assertTrue(called['power_on'])
-        self.compute.terminate_instance(self.context, instance=instance)
+        self.compute.terminate_instance(self.context, instance=inst_obj)
 
     def test_power_off(self):
         # Ensure instance can be powered off.
@@ -1357,11 +1406,15 @@ class ComputeTestCase(BaseTestCase):
 
         instance = jsonutils.to_primitive(self._create_fake_instance())
         self.compute.run_instance(self.context, instance=instance)
-        db.instance_update(self.context, instance['uuid'],
-                           {"task_state": task_states.POWERING_OFF})
-        self.compute.stop_instance(self.context, instance=instance)
+        extra = ['system_metadata', 'metadata']
+        inst_obj = instance_obj.Instance.get_by_uuid(self.context,
+                                                     instance['uuid'],
+                                                     expected_attrs=extra)
+        inst_obj.task_state = task_states.POWERING_OFF
+        inst_obj.save(self.context)
+        self.compute.stop_instance(self.context, instance=inst_obj)
         self.assertTrue(called['power_off'])
-        self.compute.terminate_instance(self.context, instance=instance)
+        self.compute.terminate_instance(self.context, instance=inst_obj)
 
     def test_pause(self):
         # Ensure instance can be paused and unpaused.
@@ -5334,47 +5387,85 @@ class ComputeAPITestCase(BaseTestCase):
             db.instance_destroy(self.context, ref[0]['uuid'])
 
     def test_start(self):
-        instance = jsonutils.to_primitive(self._create_fake_instance())
-        instance_uuid = instance['uuid']
-        self.compute.run_instance(self.context, instance=instance)
-
-        db.instance_update(self.context, instance['uuid'],
-                           {"task_state": task_states.POWERING_OFF})
-        self.compute.stop_instance(self.context, instance=instance)
+        # Undo setUp() stubs (test_compute_cells)
+        self.stubs.UnsetAll()
+        instance = self._create_instance_obj()
+        instance.vm_state = vm_states.STOPPED
+
+        self.mox.StubOutWithMock(instance, 'save')
+        self.mox.StubOutWithMock(self.compute_api,
+                '_record_action_start')
+        self.mox.StubOutWithMock(
+                self.compute_api.compute_rpcapi,
+                'start_instance')
+
+        instance.save(expected_task_state=None)
+        self.compute_api._record_action_start(self.context,
+                instance, instance_actions.START)
+        self.compute_api.compute_rpcapi.start_instance(
+                self.context, instance)
 
-        instance = db.instance_get_by_uuid(self.context, instance_uuid)
-        self.assertEqual(instance['task_state'], None)
+        self.mox.ReplayAll()
 
         self.compute_api.start(self.context, instance)
+        self.assertEqual(task_states.POWERING_ON,
+                         instance.task_state)
 
-        instance = db.instance_get_by_uuid(self.context, instance_uuid)
-        self.assertEqual(instance['task_state'], task_states.POWERING_ON)
-
-        db.instance_destroy(self.context, instance['uuid'])
+    def test_start_invalid_state(self):
+        # Undo setUp() stubs (test_compute_cells)
+        self.stubs.UnsetAll()
+        instance = self._create_instance_obj()
+        instance.vm_state = vm_states.ACTIVE
+        self.assertRaises(exception.InstanceInvalidState,
+                          self.compute_api.start,
+                          self.context, instance)
 
     def test_start_no_host(self):
-        instance = self._create_fake_instance(params={'host': ''})
-
+        # Undo setUp() stubs (test_compute_cells)
+        self.stubs.UnsetAll()
+        instance = self._create_instance_obj()
+        instance.vm_state = vm_states.STOPPED
+        instance.host = ''
         self.assertRaises(exception.InstanceNotReady,
                           self.compute_api.start,
                           self.context, instance)
 
-        db.instance_destroy(self.context, instance['uuid'])
-
     def test_stop(self):
-        instance = jsonutils.to_primitive(self._create_fake_instance())
-        instance_uuid = instance['uuid']
-        self.compute.run_instance(self.context, instance=instance)
+        # Undo setUp() stubs (test_compute_cells)
+        self.stubs.UnsetAll()
+        instance = self._create_instance_obj()
+        instance.task_state = None
+        # Make sure this gets reset
+        instance.progress = 99
+
+        self.mox.StubOutWithMock(instance, 'save')
+        self.mox.StubOutWithMock(self.compute_api,
+                '_record_action_start')
+        self.mox.StubOutWithMock(
+                self.compute_api.compute_rpcapi,
+                'stop_instance')
+
+        instance.save(expected_task_state=None)
+        self.compute_api._record_action_start(self.context,
+                instance, instance_actions.STOP)
+        self.compute_api.compute_rpcapi.stop_instance(
+                self.context, instance, cast=True)
 
-        instance = db.instance_get_by_uuid(self.context, instance_uuid)
-        self.assertEqual(instance['task_state'], None)
+        self.mox.ReplayAll()
 
         self.compute_api.stop(self.context, instance)
+        self.assertEqual(task_states.POWERING_OFF,
+                         instance.task_state)
+        self.assertEqual(0, instance.progress)
 
-        instance = db.instance_get_by_uuid(self.context, instance_uuid)
-        self.assertEqual(instance['task_state'], task_states.POWERING_OFF)
-
-        db.instance_destroy(self.context, instance['uuid'])
+    def test_stop_invalid_state(self):
+        # Undo setUp() stubs (test_compute_cells)
+        self.stubs.UnsetAll()
+        instance = self._create_instance_obj()
+        instance.vm_state = vm_states.PAUSED
+        self.assertRaises(exception.InstanceInvalidState,
+                          self.compute_api.stop,
+                          self.context, instance)
 
     def test_stop_a_stopped_inst(self):
         instance = jsonutils.to_primitive(self._create_fake_instance(
@@ -5387,46 +5478,14 @@ class ComputeAPITestCase(BaseTestCase):
         db.instance_destroy(self.context, instance['uuid'])
 
     def test_stop_no_host(self):
-        instance = self._create_fake_instance(params={'host': ''})
-
+        # Undo setUp() stubs (test_compute_cells)
+        self.stubs.UnsetAll()
+        instance = self._create_instance_obj()
+        instance.host = ''
         self.assertRaises(exception.InstanceNotReady,
                           self.compute_api.stop,
                           self.context, instance)
 
-        db.instance_destroy(self.context, instance['uuid'])
-
-    def test_start_shutdown(self):
-        def check_state(instance_uuid, power_state_, vm_state_, task_state_):
-            instance = db.instance_get_by_uuid(self.context, instance_uuid)
-            self.assertEqual(instance['power_state'], power_state_)
-            self.assertEqual(instance['vm_state'], vm_state_)
-            self.assertEqual(instance['task_state'], task_state_)
-
-        def start_check_state(instance_uuid,
-                              power_state_, vm_state_, task_state_):
-            instance = db.instance_get_by_uuid(self.context, instance_uuid)
-            self.compute_api.start(self.context, instance)
-            check_state(instance_uuid, power_state_, vm_state_, task_state_)
-
-        instance = jsonutils.to_primitive(self._create_fake_instance())
-        self.compute.run_instance(self.context, instance=instance)
-
-        check_state(instance['uuid'], power_state.RUNNING, vm_states.ACTIVE,
-                    None)
-
-        # NOTE(yamahata): emulate compute.manager._sync_power_state() that
-        # the instance is shutdown by itself
-        db.instance_update(self.context, instance['uuid'],
-                           {'power_state': power_state.NOSTATE,
-                            'vm_state': vm_states.STOPPED})
-        check_state(instance['uuid'], power_state.NOSTATE, vm_states.STOPPED,
-                    None)
-
-        start_check_state(instance['uuid'], power_state.NOSTATE,
-                          vm_states.STOPPED, task_states.POWERING_ON)
-
-        db.instance_destroy(self.context, instance['uuid'])
-
     def test_delete(self):
         instance, instance_uuid = self._run_instance(params={
                 'host': CONF.host})
diff --git a/nova/tests/compute/test_rpcapi.py b/nova/tests/compute/test_rpcapi.py
index 2c04f8060..aaf90288a 100644
--- a/nova/tests/compute/test_rpcapi.py
+++ b/nova/tests/compute/test_rpcapi.py
@@ -354,15 +354,15 @@ class ComputeRpcAPITestCase(test.TestCase):
 
     def test_start_instance(self):
         self._test_compute_api('start_instance', 'cast',
-                instance=self.fake_instance)
+                instance=self.fake_instance, version='2.29')
 
     def test_stop_instance_cast(self):
         self._test_compute_api('stop_instance', 'cast',
-                instance=self.fake_instance)
+                instance=self.fake_instance, version='2.29')
 
     def test_stop_instance_call(self):
         self._test_compute_api('stop_instance', 'call',
-                instance=self.fake_instance)
+                instance=self.fake_instance, version='2.29')
 
     def test_suspend_instance(self):
         self._test_compute_api('suspend_instance', 'cast',