merge trunk

5 files changed, 388 insertions, 184 deletions

diff --git a/nova/compute/__init__.py b/nova/compute/__init__.py
index a5df2ec1a..b94f971d1 100644
--- a/nova/compute/__init__.py
+++ b/nova/compute/__init__.py
@@ -16,17 +16,4 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
-"""
-:mod:`nova.compute` -- Compute Nodes using LibVirt
-=====================================================
-
-.. automodule:: nova.compute
-   :platform: Unix
-   :synopsis: Thin wrapper around libvirt for VM mgmt.
-.. moduleauthor:: Jesse Andrews <jesse@ansolabs.com>
-.. moduleauthor:: Devin Carlen <devin.carlen@gmail.com>
-.. moduleauthor:: Vishvananda Ishaya <vishvananda@yahoo.com>
-.. moduleauthor:: Joshua McKenty <joshua@cognition.ca>
-.. moduleauthor:: Manish Singh <yosh@gimp.org>
-.. moduleauthor:: Andy Smith <andy@anarkystic.com>
-"""
+from nova.compute.api import API
diff --git a/nova/compute/api.py b/nova/compute/api.py
index 0658044d2..adf4dbe43 100644
--- a/nova/compute/api.py
+++ b/nova/compute/api.py
@@ -17,75 +17,82 @@
 #    under the License.
 
 """
-Handles all API requests relating to instances (guest vms).
+Handles all requests relating to instances (guest vms).
 """
 
 import datetime
-import logging
 import time
 
 from nova import db
 from nova import exception
 from nova import flags
+from nova import log as logging
+from nova import network
 from nova import quota
 from nova import rpc
 from nova import utils
+from nova import volume
 from nova.compute import instance_types
 from nova.db import base
 
 FLAGS = flags.FLAGS
+LOG = logging.getLogger('nova.compute.api')
 
 
-def generate_default_hostname(internal_id):
+def generate_default_hostname(instance_id):
     """Default function to generate a hostname given an instance reference."""
-    return str(internal_id)
+    return str(instance_id)
 
 
-class ComputeAPI(base.Base):
+class API(base.Base):
     """API for interacting with the compute manager."""
 
-    def __init__(self, network_manager=None, image_service=None, **kwargs):
-        if not network_manager:
-            network_manager = utils.import_object(FLAGS.network_manager)
-        self.network_manager = network_manager
+    def __init__(self, image_service=None, network_api=None, volume_api=None,
+                 **kwargs):
         if not image_service:
             image_service = utils.import_object(FLAGS.image_service)
         self.image_service = image_service
-        super(ComputeAPI, self).__init__(**kwargs)
+        if not network_api:
+            network_api = network.API()
+        self.network_api = network_api
+        if not volume_api:
+            volume_api = volume.API()
+        self.volume_api = volume_api
+        super(API, self).__init__(**kwargs)
 
     def get_network_topic(self, context, instance_id):
         try:
-            instance = self.db.instance_get_by_internal_id(context,
-                                                           instance_id)
+            instance = self.get(context, instance_id)
         except exception.NotFound as e:
-            logging.warning("Instance %d was not found in get_network_topic",
-                            instance_id)
+            LOG.warning(_("Instance %d was not found in get_network_topic"),
+                        instance_id)
             raise e
 
         host = instance['host']
         if not host:
-            raise exception.Error("Instance %d has no host" % instance_id)
+            raise exception.Error(_("Instance %d has no host") % instance_id)
         topic = self.db.queue_get_for(context, FLAGS.compute_topic, host)
         return rpc.call(context,
                         topic,
                         {"method": "get_network_topic", "args": {'fake': 1}})
 
-    def create_instances(self, context, instance_type, image_id, min_count=1,
-                         max_count=1, kernel_id=None, ramdisk_id=None,
-                         display_name='', description='', key_name=None,
-                         key_data=None, security_group='default',
-                         user_data=None,
-                         generate_hostname=generate_default_hostname):
-        """Create the number of instances requested if quote and
+    def create(self, context, instance_type,
+               image_id, kernel_id=None, ramdisk_id=None,
+               min_count=1, max_count=1,
+               display_name='', display_description='',
+               key_name=None, key_data=None, security_group='default',
+               availability_zone=None, user_data=None,
+               generate_hostname=generate_default_hostname):
+        """Create the number of instances requested if quota and
         other arguments check out ok."""
 
-        num_instances = quota.allowed_instances(context, max_count,
-                                                instance_type)
+        type_data = instance_types.INSTANCE_TYPES[instance_type]
+        num_instances = quota.allowed_instances(context, max_count, type_data)
         if num_instances < min_count:
-            logging.warn("Quota exceeeded for %s, tried to run %s instances",
-                         context.project_id, min_count)
-            raise quota.QuotaError("Instance quota exceeded. You can only "
-                                   "run %s more instances of this type." %
+            LOG.warn(_("Quota exceeeded for %s, tried to run %s instances"),
+                     context.project_id, min_count)
+            raise quota.QuotaError(_("Instance quota exceeded. You can only "
+                                     "run %s more instances of this type.") %
                                    num_instances, "InstanceLimitExceeded")
 
         is_vpn = image_id == FLAGS.vpn_image_id
@@ -95,11 +102,11 @@ class ComputeAPI(base.Base):
                 kernel_id = image.get('kernelId', None)
             if ramdisk_id is None:
                 ramdisk_id = image.get('ramdiskId', None)
-            #No kernel and ramdisk for raw images
+            # No kernel and ramdisk for raw images
             if kernel_id == str(FLAGS.null_kernel):
                 kernel_id = None
                 ramdisk_id = None
-                logging.debug("Creating a raw instance")
+                LOG.debug(_("Creating a raw instance"))
             # Make sure we have access to kernel and ramdisk (if not raw)
             if kernel_id:
                 self.image_service.show(context, kernel_id)
@@ -123,7 +130,6 @@ class ComputeAPI(base.Base):
             key_pair = db.key_pair_get(context, context.user_id, key_name)
             key_data = key_pair['public_key']
 
-        type_data = instance_types.INSTANCE_TYPES[instance_type]
         base_options = {
             'reservation_id': utils.generate_uid('r'),
             'image_id': image_id,
@@ -138,21 +144,22 @@ class ComputeAPI(base.Base):
             'vcpus': type_data['vcpus'],
             'local_gb': type_data['local_gb'],
             'display_name': display_name,
-            'display_description': description,
+            'display_description': display_description,
             'user_data': user_data or '',
             'key_name': key_name,
-            'key_data': key_data}
+            'key_data': key_data,
+            'locked': False,
+            'availability_zone': availability_zone}
 
         elevated = context.elevated()
         instances = []
-        logging.debug(_("Going to run %s instances..."), num_instances)
+        LOG.debug(_("Going to run %s instances..."), num_instances)
         for num in range(num_instances):
             instance = dict(mac_address=utils.generate_mac(),
                             launch_index=num,
                             **base_options)
             instance = self.db.instance_create(context, instance)
             instance_id = instance['id']
-            internal_id = instance['internal_id']
 
             elevated = context.elevated()
             if not security_groups:
@@ -163,14 +170,14 @@ class ComputeAPI(base.Base):
                                                     security_group_id)
 
             # Set sane defaults if not specified
-            updates = dict(hostname=generate_hostname(internal_id))
+            updates = dict(hostname=generate_hostname(instance_id))
             if 'display_name' not in instance:
-                updates['display_name'] = "Server %s" % internal_id
+                updates['display_name'] = "Server %s" % instance_id
 
-            instance = self.update_instance(context, instance_id, **updates)
+            instance = self.update(context, instance_id, **updates)
             instances.append(instance)
 
-            logging.debug(_("Casting to scheduler for %s/%s's instance %s"),
+            LOG.debug(_("Casting to scheduler for %s/%s's instance %s"),
                           context.project_id, context.user_id, instance_id)
             rpc.cast(context,
                      FLAGS.scheduler_topic,
@@ -178,6 +185,9 @@ class ComputeAPI(base.Base):
                       "args": {"topic": FLAGS.compute_topic,
                                "instance_id": instance_id}})
 
+        for group_id in security_groups:
+            self.trigger_security_group_members_refresh(elevated, group_id)
+
         return instances
 
     def ensure_default_security_group(self, context):
@@ -197,7 +207,61 @@ class ComputeAPI(base.Base):
                       'project_id': context.project_id}
             db.security_group_create(context, values)
 
-    def update_instance(self, context, instance_id, **kwargs):
+    def trigger_security_group_rules_refresh(self, context, security_group_id):
+        """Called when a rule is added to or removed from a security_group"""
+
+        security_group = self.db.security_group_get(context, security_group_id)
+
+        hosts = set()
+        for instance in security_group['instances']:
+            if instance['host'] is not None:
+                hosts.add(instance['host'])
+
+        for host in hosts:
+            rpc.cast(context,
+                     self.db.queue_get_for(context, FLAGS.compute_topic, host),
+                     {"method": "refresh_security_group_rules",
+                      "args": {"security_group_id": security_group.id}})
+
+    def trigger_security_group_members_refresh(self, context, group_id):
+        """Called when a security group gains a new or loses a member
+
+        Sends an update request to each compute node for whom this is
+        relevant."""
+
+        # First, we get the security group rules that reference this group as
+        # the grantee..
+        security_group_rules = \
+                self.db.security_group_rule_get_by_security_group_grantee(
+                                                                     context,
+                                                                     group_id)
+
+        # ..then we distill the security groups to which they belong..
+        security_groups = set()
+        for rule in security_group_rules:
+            security_groups.add(rule['parent_group_id'])
+
+        # ..then we find the instances that are members of these groups..
+        instances = set()
+        for security_group in security_groups:
+            for instance in security_group['instances']:
+                instances.add(instance['id'])
+
+        # ...then we find the hosts where they live...
+        hosts = set()
+        for instance in instances:
+            if instance['host']:
+                hosts.add(instance['host'])
+
+        # ...and finally we tell these nodes to refresh their view of this
+        # particular security group.
+        for host in hosts:
+            rpc.cast(context,
+                     self.db.queue_get_for(context, FLAGS.compute_topic, host),
+                     {"method": "refresh_security_group_members",
+                      "args": {"security_group_id": group_id}})
+
+    def update(self, context, instance_id, **kwargs):
         """Updates the instance in the datastore.
 
         :param context: The security context
@@ -211,132 +275,138 @@ class ComputeAPI(base.Base):
         """
         return self.db.instance_update(context, instance_id, kwargs)
 
-    def delete_instance(self, context, instance_id):
-        logging.debug("Going to try and terminate %d" % instance_id)
+    def delete(self, context, instance_id):
+        LOG.debug(_("Going to try and terminate %s"), instance_id)
         try:
-            instance = self.db.instance_get_by_internal_id(context,
-                                                           instance_id)
+            instance = self.get(context, instance_id)
         except exception.NotFound as e:
-            logging.warning(_("Instance %d was not found during terminate"),
-                            instance_id)
+            LOG.warning(_("Instance %d was not found during terminate"),
+                        instance_id)
             raise e
 
         if (instance['state_description'] == 'terminating'):
-            logging.warning(_("Instance %d is already being terminated"),
-                            instance_id)
+            LOG.warning(_("Instance %d is already being terminated"),
+                        instance_id)
             return
 
-        self.update_instance(context,
-                             instance['id'],
-                             state_description='terminating',
-                             state=0,
-                             terminated_at=datetime.datetime.utcnow())
+        self.update(context,
+                    instance['id'],
+                    state_description='terminating',
+                    state=0,
+                    terminated_at=datetime.datetime.utcnow())
 
         host = instance['host']
         if host:
             rpc.cast(context,
                      self.db.queue_get_for(context, FLAGS.compute_topic, host),
                      {"method": "terminate_instance",
-                      "args": {"instance_id": instance['id']}})
+                      "args": {"instance_id": instance_id}})
         else:
-            self.db.instance_destroy(context, instance['id'])
-
-    def get_instances(self, context, project_id=None):
-        """Get all instances, possibly filtered by project ID or
-        user ID. If there is no filter and the context is an admin,
-        it will retreive all instances in the system."""
+            self.db.instance_destroy(context, instance_id)
+
+    def get(self, context, instance_id):
+        """Get a single instance with the given ID."""
+        return self.db.instance_get_by_id(context, instance_id)
+
+    def get_all(self, context, project_id=None, reservation_id=None,
+                fixed_ip=None):
+        """Get all instances, possibly filtered by one of the
+        given parameters. If there is no filter and the context is
+        an admin, it will retreive all instances in the system."""
+        if reservation_id is not None:
+            return self.db.instance_get_all_by_reservation(context,
+                                                           reservation_id)
+        if fixed_ip is not None:
+            return self.db.fixed_ip_get_instance(context, fixed_ip)
         if project_id or not context.is_admin:
             if not context.project:
                 return self.db.instance_get_all_by_user(context,
                                                         context.user_id)
             if project_id is None:
                 project_id = context.project_id
-            return self.db.instance_get_all_by_project(context, project_id)
+            return self.db.instance_get_all_by_project(context,
+            project_id)
         return self.db.instance_get_all(context)
 
-    def get_instance(self, context, instance_id):
-        return self.db.instance_get_by_internal_id(context, instance_id)
-
     def snapshot(self, context, instance_id, name):
         """Snapshot the given instance."""
-        instance = self.db.instance_get_by_internal_id(context, instance_id)
+        instance = self.get(context, instance_id)
         host = instance['host']
         rpc.cast(context,
                  self.db.queue_get_for(context, FLAGS.compute_topic, host),
                  {"method": "snapshot_instance",
-                  "args": {"instance_id": instance['id'], "name": name}})
+                  "args": {"instance_id": instance_id, "name": name}})
 
     def reboot(self, context, instance_id):
         """Reboot the given instance."""
-        instance = self.db.instance_get_by_internal_id(context, instance_id)
+        instance = self.get(context, instance_id)
         host = instance['host']
         rpc.cast(context,
                  self.db.queue_get_for(context, FLAGS.compute_topic, host),
                  {"method": "reboot_instance",
-                  "args": {"instance_id": instance['id']}})
+                  "args": {"instance_id": instance_id}})
 
     def pause(self, context, instance_id):
         """Pause the given instance."""
-        instance = self.db.instance_get_by_internal_id(context, instance_id)
+        instance = self.get(context, instance_id)
         host = instance['host']
         rpc.cast(context,
                  self.db.queue_get_for(context, FLAGS.compute_topic, host),
                  {"method": "pause_instance",
-                  "args": {"instance_id": instance['id']}})
+                  "args": {"instance_id": instance_id}})
 
     def unpause(self, context, instance_id):
         """Unpause the given instance."""
-        instance = self.db.instance_get_by_internal_id(context, instance_id)
+        instance = self.get(context, instance_id)
         host = instance['host']
         rpc.cast(context,
                  self.db.queue_get_for(context, FLAGS.compute_topic, host),
                  {"method": "unpause_instance",
-                  "args": {"instance_id": instance['id']}})
+                  "args": {"instance_id": instance_id}})
 
     def get_diagnostics(self, context, instance_id):
         """Retrieve diagnostics for the given instance."""
-        instance = self.db.instance_get_by_internal_id(context, instance_id)
+        instance = self.get(context, instance_id)
         host = instance["host"]
         return rpc.call(context,
             self.db.queue_get_for(context, FLAGS.compute_topic, host),
             {"method": "get_diagnostics",
-             "args": {"instance_id": instance["id"]}})
+             "args": {"instance_id": instance_id}})
 
     def get_actions(self, context, instance_id):
         """Retrieve actions for the given instance."""
-        instance = self.db.instance_get_by_internal_id(context, instance_id)
-        return self.db.instance_get_actions(context, instance["id"])
+        return self.db.instance_get_actions(context, instance_id)
 
     def suspend(self, context, instance_id):
         """suspend the instance with instance_id"""
-        instance = self.db.instance_get_by_internal_id(context, instance_id)
+        instance = self.get(context, instance_id)
         host = instance['host']
         rpc.cast(context,
                  self.db.queue_get_for(context, FLAGS.compute_topic, host),
                  {"method": "suspend_instance",
-                  "args": {"instance_id": instance['id']}})
+                  "args": {"instance_id": instance_id}})
 
     def resume(self, context, instance_id):
         """resume the instance with instance_id"""
-        instance = self.db.instance_get_by_internal_id(context, instance_id)
+        instance = self.get(context, instance_id)
         host = instance['host']
         rpc.cast(context,
                  self.db.queue_get_for(context, FLAGS.compute_topic, host),
                  {"method": "resume_instance",
-                  "args": {"instance_id": instance['id']}})
+                  "args": {"instance_id": instance_id}})
 
     def rescue(self, context, instance_id):
         """Rescue the given instance."""
-        instance = self.db.instance_get_by_internal_id(context, instance_id)
+        instance = self.get(context, instance_id)
         host = instance['host']
         rpc.cast(context,
                  self.db.queue_get_for(context, FLAGS.compute_topic, host),
                  {"method": "rescue_instance",
-                  "args": {"instance_id": instance['id']}})
+                  "args": {"instance_id": instance_id}})
 
     def unrescue(self, context, instance_id):
         """Unrescue the given instance."""
-        instance = self.db.instance_get_by_internal_id(context, instance_id)
+        instance = self.get(context, instance_id)
         host = instance['host']
         rpc.cast(context,
                  self.db.queue_get_for(context, FLAGS.compute_topic, host),
@@ -344,11 +414,7 @@ class ComputeAPI(base.Base):
                   "args": {"instance_id": instance['id']}})
 
     def get_ajax_console(self, context, instance_id):
-        """Get an AJAX Console
-
-        In order for this to work properly, a ttyS0 must be configured
-        in the instance
-        """
+        """Get a url to an AJAX Console"""
 
         instance_ref = db.instance_get_by_internal_id(context, instance_id)
 
@@ -366,3 +432,66 @@ class ComputeAPI(base.Base):
         return {'url': '%s?token=%s' % (FLAGS.ajax_console_proxy_url,
                 output['token'])}
 
+    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. "
+                                     "Example device: /dev/vdb") % device)
+        self.volume_api.check_attach(context, volume_id)
+        instance = self.get(context, instance_id)
+        host = instance['host']
+        rpc.cast(context,
+                 self.db.queue_get_for(context, FLAGS.compute_topic, host),
+                 {"method": "attach_volume",
+                  "args": {"volume_id": volume_id,
+                           "instance_id": instance_id,
+                           "mountpoint": device}})
+
+    def detach_volume(self, context, volume_id):
+        instance = self.db.volume_get_instance(context.elevated(), volume_id)
+        if not instance:
+            raise exception.ApiError(_("Volume isn't attached to anything!"))
+        self.volume_api.check_detach(context, volume_id)
+        host = instance['host']
+        rpc.cast(context,
+                 self.db.queue_get_for(context, FLAGS.compute_topic, host),
+                 {"method": "detach_volume",
+                  "args": {"instance_id": instance['id'],
+                           "volume_id": volume_id}})
+        return instance
+
+    def associate_floating_ip(self, context, instance_id, address):
+        instance = self.get(context, instance_id)
+        self.network_api.associate_floating_ip(context, address,
+                                               instance['fixed_ip'])
diff --git a/nova/compute/disk.py b/nova/compute/disk.py
index 814a258cd..741499294 100644
--- a/nova/compute/disk.py
+++ b/nova/compute/disk.py
@@ -22,14 +22,15 @@ Includes injection of SSH PGP keys into authorized_keys file.
 
 """
 
-import logging
 import os
 import tempfile
 
 from nova import exception
 from nova import flags
+from nova import log as logging
 
 
+LOG = logging.getLogger('nova.compute.disk')
 FLAGS = flags.FLAGS
 flags.DEFINE_integer('minimum_root_size', 1024 * 1024 * 1024 * 10,
                      'minimum size in bytes of root partition')
@@ -67,12 +68,12 @@ def partition(infile, outfile, local_bytes=0, resize=True,
         execute('resize2fs %s' % infile)
         file_size = FLAGS.minimum_root_size
     elif file_size % sector_size != 0:
-        logging.warn(_("Input partition size not evenly divisible by"
-                       " sector size: %d / %d"), file_size, sector_size)
+        LOG.warn(_("Input partition size not evenly divisible by"
+                   " sector size: %d / %d"), file_size, sector_size)
     primary_sectors = file_size / sector_size
     if local_bytes % sector_size != 0:
-        logging.warn(_("Bytes for local storage not evenly divisible"
-                       " by sector size: %d / %d"), local_bytes, sector_size)
+        LOG.warn(_("Bytes for local storage not evenly divisible"
+                   " by sector size: %d / %d"), local_bytes, sector_size)
     local_sectors = local_bytes / sector_size
 
     mbr_last = 62  # a
diff --git a/nova/compute/manager.py b/nova/compute/manager.py
index e485a0415..fd1e983fa 100644
--- a/nova/compute/manager.py
+++ b/nova/compute/manager.py
@@ -35,10 +35,11 @@ terminating it.
 """
 
 import datetime
-import logging
+import functools
 
 from nova import exception
 from nova import flags
+from nova import log as logging
 from nova import manager
 from nova import rpc
 from nova import utils
@@ -52,6 +53,42 @@ flags.DEFINE_string('compute_driver', 'nova.virt.connection.get_connection',
 flags.DEFINE_string('stub_network', False,
                     'Stub network related code')
 
+LOG = logging.getLogger('nova.compute.manager')
+
+
+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):
+
+        LOG.info(_("check_instance_lock: decorating: |%s|"), function,
+                 context=context)
+        LOG.info(_("check_instance_lock: arguments: |%s| |%s| |%s|"),
+                 self, context, instance_id, context=context)
+        locked = self.get_lock(context, instance_id)
+        admin = context.is_admin
+        LOG.info(_("check_instance_lock: locked: |%s|"), locked,
+                 context=context)
+        LOG.info(_("check_instance_lock: admin: |%s|"), admin,
+                 context=context)
+
+        # if admin or unlocked call function otherwise log error
+        if admin or not locked:
+            LOG.info(_("check_instance_lock: executing: |%s|"), function,
+                     context=context)
+            function(self, context, instance_id, *args, **kwargs)
+        else:
+            LOG.error(_("check_instance_lock: not executing |%s|"),
+                      function, context=context)
+            return False
+
+    return decorated_function
+
 
 class ComputeManager(manager.Manager):
 
@@ -100,9 +137,16 @@ class ComputeManager(manager.Manager):
                                      host)
 
     @exception.wrap_exception
-    def refresh_security_group(self, context, security_group_id, **_kwargs):
-        """This call passes stright through to the virtualization driver."""
-        self.driver.refresh_security_group(security_group_id)
+    def refresh_security_group_rules(self, context,
+                                     security_group_id, **_kwargs):
+        """This call passes straight through to the virtualization driver."""
+        return self.driver.refresh_security_group_rules(security_group_id)
+
+    @exception.wrap_exception
+    def refresh_security_group_members(self, context,
+                                       security_group_id, **_kwargs):
+        """This call passes straight through to the virtualization driver."""
+        return self.driver.refresh_security_group_members(security_group_id)
 
     @exception.wrap_exception
     def run_instance(self, context, instance_id, **_kwargs):
@@ -111,7 +155,8 @@ class ComputeManager(manager.Manager):
         instance_ref = self.db.instance_get(context, instance_id)
         if instance_ref['name'] in self.driver.list_instances():
             raise exception.Error(_("Instance has already been created"))
-        logging.debug(_("instance %s: starting..."), instance_id)
+        LOG.audit(_("instance %s: starting..."), instance_id,
+                  context=context)
         self.db.instance_update(context,
                                 instance_id,
                                 {'host': self.host})
@@ -149,8 +194,8 @@ class ComputeManager(manager.Manager):
                                     instance_id,
                                     {'launched_at': now})
         except Exception:  # pylint: disable-msg=W0702
-            logging.exception(_("instance %s: Failed to spawn"),
-                              instance_ref['name'])
+            LOG.exception(_("instance %s: Failed to spawn"), instance_id,
+                          context=context)
             self.db.instance_set_state(context,
                                        instance_id,
                                        power_state.SHUTDOWN)
@@ -158,17 +203,19 @@ 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()
-
         instance_ref = self.db.instance_get(context, instance_id)
+        LOG.audit(_("Terminating instance %s"), instance_id, context=context)
 
         if not FLAGS.stub_network:
             address = self.db.instance_get_floating_address(context,
                                                             instance_ref['id'])
             if address:
-                logging.debug(_("Disassociating address %s") % address)
+                LOG.debug(_("Disassociating address %s"), address,
+                          context=context)
                 # NOTE(vish): Right now we don't really care if the ip is
                 #             disassociated.  We may need to worry about
                 #             checking this later.
@@ -180,15 +227,14 @@ class ComputeManager(manager.Manager):
             address = self.db.instance_get_fixed_address(context,
                                                          instance_ref['id'])
             if address:
-                logging.debug(_("Deallocating address %s") % address)
+                LOG.debug(_("Deallocating address %s"), address,
+                          context=context)
                 # NOTE(vish): Currently, nothing needs to be done on the
                 #             network node until release. If this changes,
                 #             we will need to cast here.
                 self.network_manager.deallocate_fixed_ip(context.elevated(),
                                                          address)
 
-        logging.debug(_("instance %s: terminating"), instance_id)
-
         volumes = instance_ref.get('volumes', []) or []
         for volume in volumes:
             self.detach_volume(context, instance_id, volume['id'])
@@ -202,20 +248,22 @@ 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()
         self._update_state(context, instance_id)
         instance_ref = self.db.instance_get(context, instance_id)
+        LOG.audit(_("Rebooting instance %s"), instance_id, context=context)
 
         if instance_ref['state'] != power_state.RUNNING:
-            logging.warn(_('trying to reboot a non-running '
-                           'instance: %s (state: %s excepted: %s)'),
-                         instance_ref['internal_id'],
-                         instance_ref['state'],
-                         power_state.RUNNING)
+            LOG.warn(_('trying to reboot a non-running '
+                     'instance: %s (state: %s excepted: %s)'),
+                     instance_id,
+                     instance_ref['state'],
+                     power_state.RUNNING,
+                     context=context)
 
-        logging.debug(_('instance %s: rebooting'), instance_ref['name'])
         self.db.instance_set_state(context,
                                    instance_id,
                                    power_state.NOSTATE,
@@ -235,24 +283,22 @@ class ComputeManager(manager.Manager):
         # potentially?
         self._update_state(context, instance_id)
 
-        logging.debug(_('instance %s: snapshotting'), instance_ref['name'])
+        LOG.audit(_('instance %s: snapshotting'), instance_id,
+                  context=context)
         if instance_ref['state'] != power_state.RUNNING:
-            logging.warn(_('trying to snapshot a non-running '
-                           'instance: %s (state: %s excepted: %s)'),
-                         instance_ref['internal_id'],
-                         instance_ref['state'],
-                         power_state.RUNNING)
+            LOG.warn(_('trying to snapshot a non-running '
+                       'instance: %s (state: %s excepted: %s)'),
+                     instance_id, instance_ref['state'], power_state.RUNNING)
 
         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()
         instance_ref = self.db.instance_get(context, instance_id)
-
-        logging.debug(_('instance %s: rescuing'),
-                      instance_ref['internal_id'])
+        LOG.audit(_('instance %s: rescuing'), instance_id, context=context)
         self.db.instance_set_state(context,
                                    instance_id,
                                    power_state.NOSTATE,
@@ -262,13 +308,12 @@ 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()
         instance_ref = self.db.instance_get(context, instance_id)
-
-        logging.debug(_('instance %s: unrescuing'),
-                      instance_ref['internal_id'])
+        LOG.audit(_('instance %s: unrescuing'), instance_id, context=context)
         self.db.instance_set_state(context,
                                    instance_id,
                                    power_state.NOSTATE,
@@ -282,13 +327,12 @@ 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()
         instance_ref = self.db.instance_get(context, instance_id)
-
-        logging.debug('instance %s: pausing',
-                      instance_ref['internal_id'])
+        LOG.audit(_('instance %s: pausing'), instance_id, context=context)
         self.db.instance_set_state(context,
                                    instance_id,
                                    power_state.NOSTATE,
@@ -300,13 +344,12 @@ 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()
         instance_ref = self.db.instance_get(context, instance_id)
-
-        logging.debug('instance %s: unpausing',
-                      instance_ref['internal_id'])
+        LOG.audit(_('instance %s: unpausing'), instance_id, context=context)
         self.db.instance_set_state(context,
                                    instance_id,
                                    power_state.NOSTATE,
@@ -323,17 +366,20 @@ class ComputeManager(manager.Manager):
         instance_ref = self.db.instance_get(context, instance_id)
 
         if instance_ref["state"] == power_state.RUNNING:
-            logging.debug(_("instance %s: retrieving diagnostics"),
-                instance_ref["internal_id"])
+            LOG.audit(_("instance %s: retrieving diagnostics"), instance_id,
+                      context=context)
             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)
-
-        logging.debug(_('instance %s: suspending'),
-                      instance_ref['internal_id'])
+        LOG.audit(_('instance %s: suspending'), instance_id, context=context)
         self.db.instance_set_state(context, instance_id,
                                             power_state.NOSTATE,
                                             'suspending')
@@ -344,12 +390,15 @@ 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)
-
-        logging.debug(_('instance %s: resuming'), instance_ref['internal_id'])
+        LOG.audit(_('instance %s: resuming'), instance_id, context=context)
         self.db.instance_set_state(context, instance_id,
                                             power_state.NOSTATE,
                                             'resuming')
@@ -360,30 +409,66 @@ 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)
+
+        LOG.debug(_('instance %s: locking'), instance_id, context=context)
+        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)
+
+        LOG.debug(_('instance %s: unlocking'), instance_id, context=context)
+        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()
+        LOG.debug(_('instance %s: getting locked state'), instance_id,
+                  context=context)
+        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()
-        logging.debug(_("instance %s: getting console output"), instance_id)
         instance_ref = self.db.instance_get(context, instance_id)
-
+        LOG.audit(_("Get console output for instance %s"), instance_id,
+                  context=context)
         return self.driver.get_console_output(instance_ref)
 
     @exception.wrap_exception
     def get_ajax_console(self, context, instance_id):
-        """Send the console output for an instance."""
+        """Return connection information for an ajax console"""
         context = context.elevated()
         logging.debug(_("instance %s: getting ajax console"), instance_id)
         instance_ref = self.db.instance_get(context, instance_id)
 
         return self.driver.get_ajax_console(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()
-        logging.debug(_("instance %s: attaching volume %s to %s"), instance_id,
-            volume_id, mountpoint)
         instance_ref = self.db.instance_get(context, instance_id)
+        LOG.audit(_("instance %s: attaching volume %s to %s"), instance_id,
+                  volume_id, mountpoint, context=context)
         dev_path = self.volume_manager.setup_compute_volume(context,
                                                             volume_id)
         try:
@@ -398,8 +483,8 @@ class ComputeManager(manager.Manager):
             # NOTE(vish): The inline callback eats the exception info so we
             #             log the traceback here and reraise the same
             #             ecxception below.
-            logging.exception(_("instance %s: attach failed %s, removing"),
-                              instance_id, mountpoint)
+            LOG.exception(_("instance %s: attach failed %s, removing"),
+                          instance_id, mountpoint, context=context)
             self.volume_manager.remove_compute_volume(context,
                                                       volume_id)
             raise exc
@@ -407,17 +492,18 @@ 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()
-        logging.debug(_("instance %s: detaching volume %s"),
-                      instance_id,
-                      volume_id)
         instance_ref = self.db.instance_get(context, instance_id)
         volume_ref = self.db.volume_get(context, volume_id)
+        LOG.audit(_("Detach volume %s from mountpoint %s on instance %s"),
+                  volume_id, volume_ref['mountpoint'], instance_id,
+                  context=context)
         if instance_ref['name'] not in self.driver.list_instances():
-            logging.warn(_("Detaching volume from unknown instance %s"),
-                         instance_ref['name'])
+            LOG.warn(_("Detaching volume from unknown instance %s"),
+                     instance_id, context=context)
         else:
             self.driver.detach_volume(instance_ref['name'],
                                       volume_ref['mountpoint'])
diff --git a/nova/compute/monitor.py b/nova/compute/monitor.py
index 60c347a5e..14d0e8ca1 100644
--- a/nova/compute/monitor.py
+++ b/nova/compute/monitor.py
@@ -25,19 +25,17 @@ Instance Monitoring:
 """
 
 import datetime
-import logging
 import os
-import sys
 import time
 
 import boto
 import boto.s3
 import rrdtool
-from twisted.internet import defer
 from twisted.internet import task
 from twisted.application import service
 
 from nova import flags
+from nova import log as logging
 from nova.virt import connection as virt_connection
 
 
@@ -91,6 +89,9 @@ RRD_VALUES = {
 utcnow = datetime.datetime.utcnow
 
 
+LOG = logging.getLogger('nova.compute.monitor')
+
+
 def update_rrd(instance, name, data):
     """
     Updates the specified RRD file.
@@ -255,20 +256,20 @@ class Instance(object):
         Updates the instances statistics and stores the resulting graphs
         in the internal object store on the cloud controller.
         """
-        logging.debug(_('updating %s...'), self.instance_id)
+        LOG.debug(_('updating %s...'), self.instance_id)
 
         try:
             data = self.fetch_cpu_stats()
             if data != None:
-                logging.debug('CPU: %s', data)
+                LOG.debug('CPU: %s', data)
                 update_rrd(self, 'cpu', data)
 
             data = self.fetch_net_stats()
-            logging.debug('NET: %s', data)
+            LOG.debug('NET: %s', data)
             update_rrd(self, 'net', data)
 
             data = self.fetch_disk_stats()
-            logging.debug('DISK: %s', data)
+            LOG.debug('DISK: %s', data)
             update_rrd(self, 'disk', data)
 
             # TODO(devcamcar): Turn these into pool.ProcessPool.execute() calls
@@ -285,7 +286,7 @@ class Instance(object):
             graph_disk(self, '1w')
             graph_disk(self, '1m')
         except Exception:
-            logging.exception(_('unexpected error during update'))
+            LOG.exception(_('unexpected error during update'))
 
         self.last_updated = utcnow()
 
@@ -309,7 +310,7 @@ class Instance(object):
         self.cputime = float(info['cpu_time'])
         self.cputime_last_updated = utcnow()
 
-        logging.debug('CPU: %d', self.cputime)
+        LOG.debug('CPU: %d', self.cputime)
 
         # Skip calculation on first pass. Need delta to get a meaningful value.
         if cputime_last_updated == None:
@@ -319,17 +320,17 @@ class Instance(object):
         d = self.cputime_last_updated - cputime_last_updated
         t = d.days * 86400 + d.seconds
 
-        logging.debug('t = %d', t)
+        LOG.debug('t = %d', t)
 
         # Calculate change over time in number of nanoseconds of CPU time used.
         cputime_delta = self.cputime - cputime_last
 
-        logging.debug('cputime_delta = %s', cputime_delta)
+        LOG.debug('cputime_delta = %s', cputime_delta)
 
         # Get the number of virtual cpus in this domain.
         vcpus = int(info['num_cpu'])
 
-        logging.debug('vcpus = %d', vcpus)
+        LOG.debug('vcpus = %d', vcpus)
 
         # Calculate CPU % used and cap at 100.
         return min(cputime_delta / (t * vcpus * 1.0e9) * 100, 100)
@@ -351,8 +352,8 @@ class Instance(object):
                 rd += rd_bytes
                 wr += wr_bytes
             except TypeError:
-                logging.error(_('Cannot get blockstats for "%s" on "%s"'),
-                              disk, self.instance_id)
+                LOG.error(_('Cannot get blockstats for "%s" on "%s"'),
+                          disk, self.instance_id)
                 raise
 
         return '%d:%d' % (rd, wr)
@@ -373,8 +374,8 @@ class Instance(object):
                 rx += stats[0]
                 tx += stats[4]
             except TypeError:
-                logging.error(_('Cannot get ifstats for "%s" on "%s"'),
-                              interface, self.instance_id)
+                LOG.error(_('Cannot get ifstats for "%s" on "%s"'),
+                          interface, self.instance_id)
                 raise
 
         return '%d:%d' % (rx, tx)
@@ -408,7 +409,7 @@ class InstanceMonitor(object, service.Service):
         try:
             conn = virt_connection.get_connection(read_only=True)
         except Exception, exn:
-            logging.exception(_('unexpected exception getting connection'))
+            LOG.exception(_('unexpected exception getting connection'))
             time.sleep(FLAGS.monitoring_instances_delay)
             return
 
@@ -416,14 +417,14 @@ class InstanceMonitor(object, service.Service):
         try:
             self.updateInstances_(conn, domain_ids)
         except Exception, exn:
-            logging.exception('updateInstances_')
+            LOG.exception('updateInstances_')
 
     def updateInstances_(self, conn, domain_ids):
         for domain_id in domain_ids:
             if not domain_id in self._instances:
                 instance = Instance(conn, domain_id)
                 self._instances[domain_id] = instance
-                logging.debug(_('Found instance: %s'), domain_id)
+                LOG.debug(_('Found instance: %s'), domain_id)
 
         for key in self._instances.keys():
             instance = self._instances[key]