Merge from trunk

11 files changed, 211 insertions, 19 deletions

diff --git a/nova/api/ec2/cloud.py b/nova/api/ec2/cloud.py
index 0c0027287..6619b5452 100644
--- a/nova/api/ec2/cloud.py
+++ b/nova/api/ec2/cloud.py
@@ -592,7 +592,7 @@ class CloudController(object):
         return {'reservationSet': self._format_instances(context)}
 
     def _format_run_instances(self, context, reservation_id):
-        i = self._format_instances(context, reservation_id)
+        i = self._format_instances(context, reservation_id=reservation_id)
         assert len(i) == 1
         return i[0]
 
diff --git a/nova/api/openstack/servers.py b/nova/api/openstack/servers.py
index ce64ac7ad..10679ccb6 100644
--- a/nova/api/openstack/servers.py
+++ b/nova/api/openstack/servers.py
@@ -170,6 +170,50 @@ class Controller(wsgi.Controller):
             return faults.Fault(exc.HTTPUnprocessableEntity())
         return exc.HTTPAccepted()
 
+    def lock(self, req, id):
+        """
+        lock the instance with id
+        admin only operation
+
+        """
+        context = req.environ['nova.context']
+        try:
+            self.compute_api.lock(context, id)
+        except:
+            readable = traceback.format_exc()
+            logging.error(_("Compute.api::lock %s"), readable)
+            return faults.Fault(exc.HTTPUnprocessableEntity())
+        return exc.HTTPAccepted()
+
+    def unlock(self, req, id):
+        """
+        unlock the instance with id
+        admin only operation
+
+        """
+        context = req.environ['nova.context']
+        try:
+            self.compute_api.unlock(context, id)
+        except:
+            readable = traceback.format_exc()
+            logging.error(_("Compute.api::unlock %s"), readable)
+            return faults.Fault(exc.HTTPUnprocessableEntity())
+        return exc.HTTPAccepted()
+
+    def get_lock(self, req, id):
+        """
+        return the boolean state of (instance with id)'s lock
+
+        """
+        context = req.environ['nova.context']
+        try:
+            self.compute_api.get_lock(context, id)
+        except:
+            readable = traceback.format_exc()
+            logging.error(_("Compute.api::get_lock %s"), readable)
+            return faults.Fault(exc.HTTPUnprocessableEntity())
+        return exc.HTTPAccepted()
+
     def pause(self, req, id):
         """ Permit Admins to Pause the server. """
         ctxt = req.environ['nova.context']
@@ -222,4 +266,13 @@ class Controller(wsgi.Controller):
     def actions(self, req, id):
         """Permit Admins to retrieve server actions."""
         ctxt = req.environ["nova.context"]
-        return self.compute_api.get_actions(ctxt, id)
+        items = self.compute_api.get_actions(ctxt, id)
+        actions = []
+        # TODO(jk0): Do not do pre-serialization here once the default
+        # serializer is updated
+        for item in items:
+            actions.append(dict(
+                created_at=str(item.created_at),
+                action=item.action,
+                error=item.error))
+        return dict(actions=actions)
diff --git a/nova/compute/api.py b/nova/compute/api.py
index 64d47b1ce..015934ee7 100644
--- a/nova/compute/api.py
+++ b/nova/compute/api.py
@@ -147,6 +147,7 @@ class API(base.Base):
             'user_data': user_data or '',
             'key_name': key_name,
             'key_data': key_data,
+            'locked': False,
             'availability_zone': availability_zone}
 
         elevated = context.elevated()
@@ -354,6 +355,38 @@ class API(base.Base):
                  {"method": "unrescue_instance",
                   "args": {"instance_id": instance_id}})
 
+    def lock(self, context, instance_id):
+        """
+        lock the instance with instance_id
+
+        """
+        instance = self.get_instance(context, instance_id)
+        host = instance['host']
+        rpc.cast(context,
+                 self.db.queue_get_for(context, FLAGS.compute_topic, host),
+                 {"method": "lock_instance",
+                  "args": {"instance_id": instance['id']}})
+
+    def unlock(self, context, instance_id):
+        """
+        unlock the instance with instance_id
+
+        """
+        instance = self.get_instance(context, instance_id)
+        host = instance['host']
+        rpc.cast(context,
+                 self.db.queue_get_for(context, FLAGS.compute_topic, host),
+                 {"method": "unlock_instance",
+                  "args": {"instance_id": instance['id']}})
+
+    def get_lock(self, context, instance_id):
+        """
+        return the boolean state of (instance with instance_id)'s lock
+
+        """
+        instance = self.get_instance(context, instance_id)
+        return instance['locked']
+
     def attach_volume(self, context, instance_id, volume_id, device):
         if not re.match("^/dev/[a-z]d[a-z]+$", device):
             raise exception.ApiError(_("Invalid device specified: %s. "
diff --git a/nova/compute/manager.py b/nova/compute/manager.py
index ca6065890..d5136eb26 100644
--- a/nova/compute/manager.py
+++ b/nova/compute/manager.py
@@ -36,6 +36,7 @@ terminating it.
 
 import datetime
 import logging
+import functools
 
 from nova import exception
 from nova import flags
@@ -53,6 +54,38 @@ flags.DEFINE_string('stub_network', False,
                     'Stub network related code')
 
 
+def checks_instance_lock(function):
+    """
+    decorator used for preventing action against locked instances
+    unless, of course, you happen to be admin
+
+    """
+
+    @functools.wraps(function)
+    def decorated_function(self, context, instance_id, *args, **kwargs):
+
+        logging.info(_("check_instance_lock: decorating: |%s|"), function)
+        logging.info(_("check_instance_lock: arguments: |%s| |%s| |%s|"),
+                                                                   self,
+                                                                   context,
+                                                                   instance_id)
+        locked = self.get_lock(context, instance_id)
+        admin = context.is_admin
+        logging.info(_("check_instance_lock: locked: |%s|"), locked)
+        logging.info(_("check_instance_lock: admin: |%s|"), admin)
+
+        # if admin or unlocked call function otherwise log error
+        if admin or not locked:
+            logging.info(_("check_instance_lock: executing: |%s|"), function)
+            function(self, context, instance_id, *args, **kwargs)
+        else:
+            logging.error(_("check_instance_lock: not executing |%s|"),
+                                                              function)
+            return False
+
+    return decorated_function
+
+
 class ComputeManager(manager.Manager):
 
     """Manages the running instances from creation to destruction."""
@@ -158,6 +191,7 @@ class ComputeManager(manager.Manager):
         self._update_state(context, instance_id)
 
     @exception.wrap_exception
+    @checks_instance_lock
     def terminate_instance(self, context, instance_id):
         """Terminate an instance on this machine."""
         context = context.elevated()
@@ -202,6 +236,7 @@ class ComputeManager(manager.Manager):
         self.db.instance_destroy(context, instance_id)
 
     @exception.wrap_exception
+    @checks_instance_lock
     def reboot_instance(self, context, instance_id):
         """Reboot an instance on this server."""
         context = context.elevated()
@@ -246,6 +281,7 @@ class ComputeManager(manager.Manager):
         self.driver.snapshot(instance_ref, name)
 
     @exception.wrap_exception
+    @checks_instance_lock
     def rescue_instance(self, context, instance_id):
         """Rescue an instance on this server."""
         context = context.elevated()
@@ -261,6 +297,7 @@ class ComputeManager(manager.Manager):
         self._update_state(context, instance_id)
 
     @exception.wrap_exception
+    @checks_instance_lock
     def unrescue_instance(self, context, instance_id):
         """Rescue an instance on this server."""
         context = context.elevated()
@@ -280,6 +317,7 @@ class ComputeManager(manager.Manager):
         self._update_state(context, instance_id)
 
     @exception.wrap_exception
+    @checks_instance_lock
     def pause_instance(self, context, instance_id):
         """Pause an instance on this server."""
         context = context.elevated()
@@ -297,6 +335,7 @@ class ComputeManager(manager.Manager):
                                                        result))
 
     @exception.wrap_exception
+    @checks_instance_lock
     def unpause_instance(self, context, instance_id):
         """Unpause a paused instance on this server."""
         context = context.elevated()
@@ -324,8 +363,12 @@ class ComputeManager(manager.Manager):
             return self.driver.get_diagnostics(instance_ref)
 
     @exception.wrap_exception
+    @checks_instance_lock
     def suspend_instance(self, context, instance_id):
-        """suspend the instance with instance_id"""
+        """
+        suspend the instance with instance_id
+
+        """
         context = context.elevated()
         instance_ref = self.db.instance_get(context, instance_id)
 
@@ -340,8 +383,12 @@ class ComputeManager(manager.Manager):
                                                        result))
 
     @exception.wrap_exception
+    @checks_instance_lock
     def resume_instance(self, context, instance_id):
-        """resume the suspended instance with instance_id"""
+        """
+        resume the suspended instance with instance_id
+
+        """
         context = context.elevated()
         instance_ref = self.db.instance_get(context, instance_id)
 
@@ -356,6 +403,41 @@ class ComputeManager(manager.Manager):
                                                        result))
 
     @exception.wrap_exception
+    def lock_instance(self, context, instance_id):
+        """
+        lock the instance with instance_id
+
+        """
+        context = context.elevated()
+        instance_ref = self.db.instance_get(context, instance_id)
+
+        logging.debug(_('instance %s: locking'), instance_id)
+        self.db.instance_update(context, instance_id, {'locked': True})
+
+    @exception.wrap_exception
+    def unlock_instance(self, context, instance_id):
+        """
+        unlock the instance with instance_id
+
+        """
+        context = context.elevated()
+        instance_ref = self.db.instance_get(context, instance_id)
+
+        logging.debug(_('instance %s: unlocking'), instance_id)
+        self.db.instance_update(context, instance_id, {'locked': False})
+
+    @exception.wrap_exception
+    def get_lock(self, context, instance_id):
+        """
+        return the boolean state of (instance with instance_id)'s lock
+
+        """
+        context = context.elevated()
+        logging.debug(_('instance %s: getting locked state'), instance_id)
+        instance_ref = self.db.instance_get(context, instance_id)
+        return instance_ref['locked']
+
+    @exception.wrap_exception
     def get_console_output(self, context, instance_id):
         """Send the console output for an instance."""
         context = context.elevated()
@@ -365,6 +447,7 @@ class ComputeManager(manager.Manager):
         return self.driver.get_console_output(instance_ref)
 
     @exception.wrap_exception
+    @checks_instance_lock
     def attach_volume(self, context, instance_id, volume_id, mountpoint):
         """Attach a volume to an instance."""
         context = context.elevated()
@@ -394,6 +477,7 @@ class ComputeManager(manager.Manager):
         return True
 
     @exception.wrap_exception
+    @checks_instance_lock
     def detach_volume(self, context, instance_id, volume_id):
         """Detach a volume from an instance."""
         context = context.elevated()
diff --git a/nova/db/sqlalchemy/__init__.py b/nova/db/sqlalchemy/__init__.py
index 22aa1cfe6..367fdda8b 100644
--- a/nova/db/sqlalchemy/__init__.py
+++ b/nova/db/sqlalchemy/__init__.py
@@ -39,5 +39,6 @@ for i in xrange(FLAGS.sql_max_retries):
         models.register_models()
         break
     except OperationalError:
-        logging.exception(_("Data store is unreachable."
-            " Trying again in %d seconds.") % FLAGS.sql_retry_interval)
+        logging.exception(_("Data store %s is unreachable."
+                            " Trying again in %d seconds.") %
+                          (FLAGS.sql_connection, FLAGS.sql_retry_interval))
diff --git a/nova/db/sqlalchemy/api.py b/nova/db/sqlalchemy/api.py
index aaa07e3c9..45427597a 100644
--- a/nova/db/sqlalchemy/api.py
+++ b/nova/db/sqlalchemy/api.py
@@ -840,12 +840,9 @@ def instance_action_create(context, values):
 def instance_get_actions(context, instance_id):
     """Return the actions associated to the given instance id"""
     session = get_session()
-    actions = {}
-    for action in session.query(models.InstanceActions).\
+    return session.query(models.InstanceActions).\
         filter_by(instance_id=instance_id).\
-        all():
-        actions[action.action] = action.error
-    return actions
+        all()
 
 
 ###################
diff --git a/nova/db/sqlalchemy/models.py b/nova/db/sqlalchemy/models.py
index 62bb1780d..1ed366127 100644
--- a/nova/db/sqlalchemy/models.py
+++ b/nova/db/sqlalchemy/models.py
@@ -224,6 +224,8 @@ class Instance(BASE, NovaBase):
     display_name = Column(String(255))
     display_description = Column(String(255))
 
+    locked = Column(Boolean)
+
     # TODO(vish): see Ewan's email about state improvements, probably
     #             should be in a driver base class or some such
     # vmstate_state = running, halted, suspended, paused
diff --git a/nova/rpc.py b/nova/rpc.py
index 844088348..ae5e77921 100644
--- a/nova/rpc.py
+++ b/nova/rpc.py
@@ -193,6 +193,7 @@ class AdapterConsumer(TopicConsumer):
             if msg_id:
                 msg_reply(msg_id, rval, None)
         except Exception as e:
+            logging.exception("Exception during message handling")
             if msg_id:
                 msg_reply(msg_id, None, sys.exc_info())
         return
diff --git a/nova/service.py b/nova/service.py
index 7203430c6..a48d00e45 100644
--- a/nova/service.py
+++ b/nova/service.py
@@ -211,9 +211,10 @@ class Service(object):
                 try:
                     models.register_models()
                 except OperationalError:
-                    logging.exception(_("Data store is unreachable."
-                        " Trying again in %d seconds.") %
-                            FLAGS.sql_retry_interval)
+                    logging.exception(_("Data store %s is unreachable."
+                                        " Trying again in %d seconds.") %
+                                      (FLAGS.sql_connection,
+                                       FLAGS.sql_retry_interval))
                     time.sleep(FLAGS.sql_retry_interval)
 
 
diff --git a/nova/tests/test_cloud.py b/nova/tests/test_cloud.py
index 42344af1c..ba58fab59 100644
--- a/nova/tests/test_cloud.py
+++ b/nova/tests/test_cloud.py
@@ -140,15 +140,16 @@ class CloudTestCase(test.TestCase):
         kwargs = {'image_id': image_id,
                   'instance_type': instance_type,
                   'max_count': max_count}
-        rv = yield self.cloud.run_instances(self.context, **kwargs)
+        rv = self.cloud.run_instances(self.context, **kwargs)
+        print rv
         instance_id = rv['instancesSet'][0]['instanceId']
-        output = yield self.cloud.get_console_output(context=self.context,
+        output = self.cloud.get_console_output(context=self.context,
                                                      instance_id=[instance_id])
         self.assertEquals(b64decode(output['output']), 'FAKE CONSOLE OUTPUT')
         # TODO(soren): We need this until we can stop polling in the rpc code
         #              for unit tests.
         greenthread.sleep(0.3)
-        rv = yield self.cloud.terminate_instances(self.context, [instance_id])
+        rv = self.cloud.terminate_instances(self.context, [instance_id])
 
     def test_key_generation(self):
         result = self._create_key('test')
@@ -186,7 +187,7 @@ class CloudTestCase(test.TestCase):
         kwargs = {'image_id': image_id,
                   'instance_type': instance_type,
                   'max_count': max_count}
-        rv = yield self.cloud.run_instances(self.context, **kwargs)
+        rv = self.cloud.run_instances(self.context, **kwargs)
         # TODO: check for proper response
         instance_id = rv['reservationSet'][0].keys()[0]
         instance = rv['reservationSet'][0][instance_id][0]
@@ -209,7 +210,7 @@ class CloudTestCase(test.TestCase):
             for instance in reservations[reservations.keys()[0]]:
                 instance_id = instance['instance_id']
                 logging.debug("Terminating instance %s" % instance_id)
-                rv = yield self.compute.terminate_instance(instance_id)
+                rv = self.compute.terminate_instance(instance_id)
 
     def test_instance_update_state(self):
         def instance(num):
diff --git a/nova/tests/test_compute.py b/nova/tests/test_compute.py
index 1d527b8f0..534493dfe 100644
--- a/nova/tests/test_compute.py
+++ b/nova/tests/test_compute.py
@@ -178,3 +178,22 @@ class ComputeTestCase(test.TestCase):
                           self.context,
                           instance_id)
         self.compute.terminate_instance(self.context, instance_id)
+
+    def test_lock(self):
+        """ensure locked instance cannot be changed"""
+        instance_id = self._create_instance()
+        self.compute.run_instance(self.context, instance_id)
+
+        non_admin_context = context.RequestContext(None, None, False, False)
+
+        # decorator should return False (fail) with locked nonadmin context
+        self.compute.lock_instance(self.context, instance_id)
+        ret_val = self.compute.reboot_instance(non_admin_context, instance_id)
+        self.assertEqual(ret_val, False)
+
+        # decorator should return None (success) with unlocked nonadmin context
+        self.compute.unlock_instance(self.context, instance_id)
+        ret_val = self.compute.reboot_instance(non_admin_context, instance_id)
+        self.assertEqual(ret_val, None)
+
+        self.compute.terminate_instance(self.context, instance_id)