diff --git a/nova/db/api.py b/nova/db/api.py
index d6eb43146098..687e7b91ba77 100644
--- a/nova/db/api.py
+++ b/nova/db/api.py
@@ -1488,6 +1488,25 @@ def instance_metadata_update(context, instance_id, metadata, delete):
 ####################
 
 
+def instance_system_metadata_get(context, instance_uuid):
+    """Get all system metadata for an instance."""
+    return IMPL.instance_system_metadata_get(context, instance_uuid)
+
+
+def instance_system_metadata_delete(context, instance_uuid, key):
+    """Delete the given system metadata item."""
+    IMPL.instance_system_metadata_delete(context, instance_uuid, key)
+
+
+def instance_system_metadata_update(context, instance_uuid, metadata, delete):
+    """Update metadata if it exists, otherwise create it."""
+    IMPL.instance_system_metadata_update(
+            context, instance_uuid, metadata, delete)
+
+
+####################
+
+
 def agent_build_create(context, values):
     """Create a new agent build entry."""
     return IMPL.agent_build_create(context, values)
diff --git a/nova/db/sqlalchemy/api.py b/nova/db/sqlalchemy/api.py
index 15a843306158..5485e7a32cf3 100644
--- a/nova/db/sqlalchemy/api.py
+++ b/nova/db/sqlalchemy/api.py
@@ -136,11 +136,25 @@ def require_instance_exists(f):
     Requires the wrapped function to use context and instance_id as
     their first two arguments.
     """
-
+    @functools.wraps(f)
     def wrapper(context, instance_id, *args, **kwargs):
         db.instance_get(context, instance_id)
         return f(context, instance_id, *args, **kwargs)
-    wrapper.__name__ = f.__name__
+
+    return wrapper
+
+
+def require_instance_exists_using_uuid(f):
+    """Decorator to require the specified instance to exist.
+
+    Requires the wrapped function to use context and instance_uuid as
+    their first two arguments.
+    """
+    @functools.wraps(f)
+    def wrapper(context, instance_uuid, *args, **kwargs):
+        db.instance_get_by_uuid(context, instance_uuid)
+        return f(context, instance_uuid, *args, **kwargs)
+
     return wrapper
 
 
@@ -1259,8 +1273,12 @@ def instance_create(context, values):
     values - dict containing column values.
     """
     values = values.copy()
-    values['metadata'] = _metadata_refs(values.get('metadata'),
-                                        models.InstanceMetadata)
+    values['metadata'] = _metadata_refs(
+            values.get('metadata'), models.InstanceMetadata)
+
+    values['system_metadata'] = _metadata_refs(
+            values.get('system_metadata'), models.InstanceSystemMetadata)
+
     instance_ref = models.Instance()
     if not values.get('uuid'):
         values['uuid'] = str(utils.gen_uuid())
@@ -1620,10 +1638,15 @@ def instance_update(context, instance_id, values):
 
     metadata = values.get('metadata')
     if metadata is not None:
-        instance_metadata_update(context,
-                                 instance_ref['id'],
-                                 values.pop('metadata'),
-                                 delete=True)
+        instance_metadata_update(
+            context, instance_ref['id'], values.pop('metadata'), delete=True)
+
+    system_metadata = values.get('system_metadata')
+    if system_metadata is not None:
+        instance_system_metadata_update(
+             context, instance_ref['uuid'], values.pop('system_metadata'),
+             delete=True)
+
     with session.begin():
         instance_ref.update(values)
         instance_ref.save(session=session)
@@ -3682,8 +3705,8 @@ def cell_get_all(context):
     return model_query(context, models.Cell, read_deleted="no").all()
 
 
-####################
-
+########################
+# User-provided metadata
 
 def _instance_metadata_get_query(context, instance_id, session=None):
     return model_query(context, models.InstanceMetadata, session=session,
@@ -3764,6 +3787,88 @@ def instance_metadata_update(context, instance_id, metadata, delete):
     return metadata
 
 
+#######################
+# System-owned metadata
+
+def _instance_system_metadata_get_query(context, instance_uuid, session=None):
+    return model_query(context, models.InstanceSystemMetadata, session=session,
+                       read_deleted="no").\
+                    filter_by(instance_uuid=instance_uuid)
+
+
+@require_context
+@require_instance_exists_using_uuid
+def instance_system_metadata_get(context, instance_uuid):
+    rows = _instance_system_metadata_get_query(context, instance_uuid).all()
+
+    result = {}
+    for row in rows:
+        result[row['key']] = row['value']
+
+    return result
+
+
+@require_context
+@require_instance_exists_using_uuid
+def instance_system_metadata_delete(context, instance_uuid, key):
+    _instance_system_metadata_get_query(context, instance_uuid).\
+        filter_by(key=key).\
+        update({'deleted': True,
+                'deleted_at': utils.utcnow(),
+                'updated_at': literal_column('updated_at')})
+
+
+def _instance_system_metadata_get_item(context, instance_uuid, key,
+                                       session=None):
+    result = _instance_system_metadata_get_query(
+                            context, instance_uuid, session=session).\
+                    filter_by(key=key).\
+                    first()
+
+    if not result:
+        raise exception.InstanceSystemMetadataNotFound(
+                metadata_key=key, instance_uuid=instance_uuid)
+
+    return result
+
+
+@require_context
+@require_instance_exists_using_uuid
+def instance_system_metadata_update(context, instance_uuid, metadata, delete):
+    session = get_session()
+
+    # Set existing metadata to deleted if delete argument is True
+    if delete:
+        original_metadata = instance_system_metadata_get(
+                context, instance_uuid)
+        for meta_key, meta_value in original_metadata.iteritems():
+            if meta_key not in metadata:
+                meta_ref = _instance_system_metadata_get_item(
+                        context, instance_uuid, meta_key, session)
+                meta_ref.update({'deleted': True})
+                meta_ref.save(session=session)
+
+    meta_ref = None
+
+    # Now update all existing items with new values, or create new meta objects
+    for meta_key, meta_value in metadata.iteritems():
+
+        # update the value whether it exists or not
+        item = {"value": meta_value}
+
+        try:
+            meta_ref = _instance_system_metadata_get_item(
+                    context, instance_uuid, meta_key, session)
+        except exception.InstanceSystemMetadataNotFound, e:
+            meta_ref = models.InstanceSystemMetadata()
+            item.update({"key": meta_key, "instance_uuid": instance_uuid})
+
+        meta_ref.update(item)
+        meta_ref.save(session=session)
+
+    return metadata
+
+
 ####################
 
 
diff --git a/nova/db/sqlalchemy/migrate_repo/versions/092_add_instance_system_metadata.py b/nova/db/sqlalchemy/migrate_repo/versions/092_add_instance_system_metadata.py
new file mode 100644
index 000000000000..dd9cb6f80158
--- /dev/null
+++ b/nova/db/sqlalchemy/migrate_repo/versions/092_add_instance_system_metadata.py
@@ -0,0 +1,71 @@
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright 2012 Openstack, LLC.
+# All Rights Reserved.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+from sqlalchemy import Boolean, Column, DateTime, ForeignKey, Integer
+from sqlalchemy import MetaData, String, Table
+from nova import log as logging
+
+LOG = logging.getLogger(__name__)
+
+
+def upgrade(migrate_engine):
+    # Upgrade operations go here. Don't create your own engine;
+    # bind migrate_engine to your metadata
+    meta = MetaData()
+    meta.bind = migrate_engine
+
+    # load tables for fk
+    instances = Table('instances', meta, autoload=True)
+
+    instance_system_metadata = Table('instance_system_metadata', meta,
+            Column('created_at', DateTime(timezone=False)),
+            Column('updated_at', DateTime(timezone=False)),
+            Column('deleted_at', DateTime(timezone=False)),
+            Column('deleted', Boolean(create_constraint=True, name=None)),
+            Column('id', Integer(), primary_key=True, nullable=False),
+            Column('instance_uuid',
+                   String(36),
+                   ForeignKey('instances.uuid'),
+                   nullable=False),
+            Column('key',
+                   String(length=255, convert_unicode=True,
+                          assert_unicode=None,
+                          unicode_error=None, _warn_on_bytestring=False),
+                   nullable=False),
+            Column('value',
+                   String(length=255, convert_unicode=True,
+                          assert_unicode=None,
+                          unicode_error=None, _warn_on_bytestring=False)),
+            mysql_engine='InnoDB')
+
+    try:
+        instance_system_metadata.create()
+    except Exception:
+        LOG.error(_("Table |%s| not created!"), repr(instance_system_metadata))
+        raise
+
+
+def downgrade(migrate_engine):
+    meta = MetaData()
+    meta.bind = migrate_engine
+
+    # load tables for fk
+    instances = Table('instances', meta, autoload=True)
+
+    instance_system_metadata = Table(
+            'instance_system_metadata', meta, autoload=True)
+    instance_system_metadata.drop()
diff --git a/nova/db/sqlalchemy/models.py b/nova/db/sqlalchemy/models.py
index e4e47c882e80..e35a78257a1e 100644
--- a/nova/db/sqlalchemy/models.py
+++ b/nova/db/sqlalchemy/models.py
@@ -831,7 +831,7 @@ class Console(BASE, NovaBase):
 
 
 class InstanceMetadata(BASE, NovaBase):
-    """Represents a metadata key/value pair for an instance"""
+    """Represents a user-provided metadata key/value pair for an instance"""
     __tablename__ = 'instance_metadata'
     id = Column(Integer, primary_key=True)
     key = Column(String(255))
@@ -844,6 +844,23 @@ class InstanceMetadata(BASE, NovaBase):
                                 'InstanceMetadata.deleted == False)')
 
 
+class InstanceSystemMetadata(BASE, NovaBase):
+    """Represents a system-owned metadata key/value pair for an instance"""
+    __tablename__ = 'instance_system_metadata'
+    id = Column(Integer, primary_key=True)
+    key = Column(String(255))
+    value = Column(String(255))
+    instance_uuid = Column(String(36),
+                           ForeignKey('instances.uuid'),
+                           nullable=False)
+
+    primary_join = ('and_(InstanceSystemMetadata.instance_uuid == '
+                    'Instance.uuid, InstanceSystemMetadata.deleted == False)')
+    instance = relationship(Instance, backref="system_metadata",
+                            foreign_keys=instance_uuid,
+                            primaryjoin=primary_join)
+
+
 class InstanceTypeExtraSpecs(BASE, NovaBase):
     """Represents additional specs as key/value pairs for an instance_type"""
     __tablename__ = 'instance_type_extra_specs'
diff --git a/nova/exception.py b/nova/exception.py
index dc1199be3d49..df2f79e3d99f 100644
--- a/nova/exception.py
+++ b/nova/exception.py
@@ -811,6 +811,11 @@ class InstanceMetadataNotFound(NotFound):
                 "key %(metadata_key)s.")
 
 
+class InstanceSystemMetadataNotFound(NotFound):
+    message = _("Instance %(instance_uuid)s has no system metadata with "
+                "key %(metadata_key)s.")
+
+
 class InstanceTypeExtraSpecsNotFound(NotFound):
     message = _("Instance Type %(instance_type_id)s has no extra specs with "
                 "key %(extra_specs_key)s.")
diff --git a/nova/tests/test_db_api.py b/nova/tests/test_db_api.py
index b5607cee47d5..c1002d98ad7f 100644
--- a/nova/tests/test_db_api.py
+++ b/nova/tests/test_db_api.py
@@ -168,37 +168,47 @@ class DbApiTestCase(test.TestCase):
         ctxt = context.get_admin_context()
 
         # Create an instance with some metadata
-        metadata = {'host': 'foo'}
-        values = {'metadata': metadata}
+        values = {'metadata': {'host': 'foo'},
+                  'system_metadata': {'original_image_ref': 'blah'}}
         instance = db.instance_create(ctxt, values)
 
         # Update the metadata
-        metadata = {'host': 'bar'}
-        values = {'metadata': metadata}
+        values = {'metadata': {'host': 'bar'},
+                  'system_metadata': {'original_image_ref': 'baz'}}
         db.instance_update(ctxt, instance.id, values)
 
-        # Retrieve the metadata to ensure it was successfully updated
+        # Retrieve the user-provided metadata to ensure it was successfully
+        # updated
         instance_meta = db.instance_metadata_get(ctxt, instance.id)
         self.assertEqual('bar', instance_meta['host'])
 
+        # Retrieve the system metadata to ensure it was successfully updated
+        system_meta = db.instance_system_metadata_get(ctxt, instance.uuid)
+        self.assertEqual('baz', system_meta['original_image_ref'])
+
     def test_instance_update_with_instance_uuid(self):
         """ test instance_update() works when an instance UUID is passed """
         ctxt = context.get_admin_context()
 
         # Create an instance with some metadata
-        metadata = {'host': 'foo'}
-        values = {'metadata': metadata}
+        values = {'metadata': {'host': 'foo'},
+                  'system_metadata': {'original_image_ref': 'blah'}}
         instance = db.instance_create(ctxt, values)
 
         # Update the metadata
-        metadata = {'host': 'bar'}
-        values = {'metadata': metadata}
+        values = {'metadata': {'host': 'bar'},
+                  'system_metadata': {'original_image_ref': 'baz'}}
         db.instance_update(ctxt, instance.uuid, values)
 
-        # Retrieve the metadata to ensure it was successfully updated
+        # Retrieve the user-provided metadata to ensure it was successfully
+        # updated
         instance_meta = db.instance_metadata_get(ctxt, instance.id)
         self.assertEqual('bar', instance_meta['host'])
 
+        # Retrieve the system metadata to ensure it was successfully updated
+        system_meta = db.instance_system_metadata_get(ctxt, instance.uuid)
+        self.assertEqual('baz', system_meta['original_image_ref'])
+
     def test_instance_fault_create(self):
         """Ensure we can create an instance fault"""
         ctxt = context.get_admin_context()