Merge "Sync SQLAlchemy lifecycle to request cycle"

barbican/api/app.py

@@ -36,6 +36,7 @@ from barbican.api.controllers import secrets
 from barbican.api.controllers import transportkeys
 from barbican.api.controllers import versions
 from barbican.common import config
+from barbican.model import repositories
 from barbican.openstack.common import log
 from barbican import queue
 
@@ -63,7 +64,17 @@ class PecanAPI(pecan.Pecan):
     performance_controller = performance.PerformanceController()
 
     def __init__(self, *args, **kwargs):
-        kwargs.setdefault('hooks', []).append(JSONErrorHook())
+        hooks = [JSONErrorHook()]
+        if kwargs.pop('is_transactional', None):
+            transaction_hook = pecan.hooks.TransactionHook(
+                repositories.start,
+                repositories.start_read_only,
+                repositories.commit,
+                repositories.rollback,
+                repositories.clear
+            )
+            hooks.append(transaction_hook)
+        kwargs['hooks'] = hooks
         super(PecanAPI, self).__init__(*args, **kwargs)
 
     def route(self, req, node, path):
@@ -95,7 +106,7 @@ def create_main_app(global_config, **local_conf):
 
     # Queuing initialization
     CONF = cfg.CONF
-    queue.init(CONF)
+    queue.init(CONF, is_server_side=False)
 
     class RootController(object):
         secrets = secrets.SecretsController()
@@ -103,7 +114,8 @@ def create_main_app(global_config, **local_conf):
         containers = containers.ContainersController()
         transport_keys = transportkeys.TransportKeysController()
 
-    wsgi_app = PecanAPI(RootController(), force_canonical=False)
+    wsgi_app = PecanAPI(
+        RootController(), is_transactional=True, force_canonical=False)
     if newrelic_loaded:
         wsgi_app = newrelic.agent.WSGIApplicationWrapper(wsgi_app)
     return wsgi_app

barbican/model/repositories.py

@@ -72,6 +72,51 @@ _CONNECTION = None
 _IDLE_TIMEOUT = None
 
 
+def start():
+    """Start database and establish a read/write connection to it.
+
+    Typically performed at the start of a request cycle, say for POST or PUT
+    requests.
+    """
+    configure_db()
+    get_session()
+
+
+def start_read_only():
+    """Start database and establish a read-only connection to it.
+
+    Typically performed at the start of a request cycle, say for GET or HEAD
+    requests.
+    """
+    # TODO(john-wood-w) Add optional, separate engine/connection for reads.
+    start()
+
+
+def commit():
+    """Commit session state so far to the database.
+
+    Typically performed at the end of a request cycle.
+    """
+    get_session().commit()
+
+
+def rollback():
+    """Rollback session state so far.
+
+    Typically performed when the request cycle raises an Exception.
+    """
+    get_session().rollback()
+
+
+def clear():
+    """Dispose of this session, releases database resources.
+
+    Typically performed at the end of a request cycle, after a
+    commit() or rollback().
+    """
+    _MAKER.remove()
+
+
 def setup_db_env():
     """Setup configuration for database."""
     global sa_logger, _IDLE_TIMEOUT, _MAX_RETRIES, _RETRY_INTERVAL, _CONNECTION
@@ -96,12 +141,12 @@ def configure_db():
     get_engine()
 
 
-def get_session(autocommit=True, expire_on_commit=False):
+def get_session():
     """Helper method to grab session."""
     global _MAKER
     if not _MAKER:
         get_engine()
-        get_maker(autocommit, expire_on_commit)
+        get_maker()
         assert(_MAKER)
     session = _MAKER()
     return session
@@ -167,7 +212,7 @@ def get_engine():
     return _ENGINE
 
 
-def get_maker(autocommit=True, expire_on_commit=False):
+def get_maker():
     """Return a SQLAlchemy sessionmaker."""
     """May assign __MAKER if not already assigned"""
     global _MAKER, _ENGINE
@@ -176,9 +221,7 @@ def get_maker(autocommit=True, expire_on_commit=False):
         # Utilize SQLAlchemy's scoped_session to ensure that we only have one
         #   session instance per thread.
         _MAKER = sqlalchemy.orm.scoped_session(
-            sa_orm.sessionmaker(bind=_ENGINE,
-                                autocommit=autocommit,
-                                expire_on_commit=expire_on_commit))
+            sa_orm.sessionmaker(bind=_ENGINE))
     return _MAKER
 
 
@@ -296,10 +339,6 @@ class BaseRepo(object):
     configuration.
     """
 
-    def __init__(self):
-        LOG.debug("BaseRepo init...")
-        configure_db()
-
     def get_session(self, session=None):
         LOG.debug("Getting session...")
         return session or get_session()
@@ -309,6 +348,7 @@ class BaseRepo(object):
             suppress_exception=False, session=None):
         """Get an entity or raise if it does not exist."""
         session = self.get_session(session)
+
         try:
             query = self._do_build_get_query(entity_id,
                                              keystone_id, session)
@@ -342,27 +382,26 @@ class BaseRepo(object):
             raise exception.Invalid(msg)
 
         LOG.debug("Begin create from...")
-        session = get_session(session)
-        with session.begin():
 
-            # Validate the attributes before we go any further. From my
-            # (unknown Glance developer) investigation, the @validates
-            # decorator does not validate
-            # on new records, only on existing records, which is, well,
-            # idiotic.
-            values = self._do_validate(entity.to_dict())
+        # Validate the attributes before we go any further. From my
+        # (unknown Glance developer) investigation, the @validates
+        # decorator does not validate
+        # on new records, only on existing records, which is, well,
+        # idiotic.
+        values = self._do_validate(entity.to_dict())
+
+        try:
+            LOG.debug("Saving entity...")
+            entity.save(session=session)
+        except sqlalchemy.exc.IntegrityError:
+            LOG.exception('Problem saving entity for create')
+            if values:
+                values_id = values['id']
+            else:
+                values_id = None
+            raise exception.Duplicate("Entity ID {0} already exists!"
+                                      .format(values_id))
 
-            try:
-                LOG.debug("Saving entity...")
-                entity.save(session=session)
-            except sqlalchemy.exc.IntegrityError:
-                LOG.exception('Problem saving entity for create')
-                if values:
-                    values_id = values['id']
-                else:
-                    values_id = None
-                raise exception.Duplicate("Entity ID {0} already exists!"
-                                          .format(values_id))
         LOG.debug('Elapsed repo '
                   'create secret:%s', (time.time() - start))  # DEBUG
 
@@ -373,23 +412,21 @@ class BaseRepo(object):
 
         :raises NotFound if entity does not exist.
         """
-        session = get_session()
-        with session.begin():
-            entity.updated_at = timeutils.utcnow()
+        entity.updated_at = timeutils.utcnow()
 
-            # Validate the attributes before we go any further. From my
-            # (unknown Glance developer) investigation, the @validates
-            # decorator does not validate
-            # on new records, only on existing records, which is, well,
-            # idiotic.
-            self._do_validate(entity.to_dict())
+        # Validate the attributes before we go any further. From my
+        # (unknown Glance developer) investigation, the @validates
+        # decorator does not validate
+        # on new records, only on existing records, which is, well,
+        # idiotic.
+        self._do_validate(entity.to_dict())
 
-            try:
-                entity.save(session=session)
-            except sqlalchemy.exc.IntegrityError:
-                LOG.exception('Problem saving entity for update')
-                raise exception.NotFound("Entity ID %s not found"
-                                         % entity.id)
+        try:
+            entity.save()
+        except sqlalchemy.exc.IntegrityError:
+            LOG.exception('Problem saving entity for update')
+            raise exception.NotFound("Entity ID %s not found"
+                                     % entity.id)
 
     def update(self, entity_id, values, purge_props=False):
         """Set the given properties on an entity and update it.
@@ -402,17 +439,16 @@ class BaseRepo(object):
         """Remove the entity by its ID."""
 
         session = get_session()
-        with session.begin():
 
-            entity = self.get(entity_id=entity_id, keystone_id=keystone_id,
-                              session=session)
+        entity = self.get(entity_id=entity_id, keystone_id=keystone_id,
+                          session=session)
 
-            try:
-                entity.delete(session=session)
-            except sqlalchemy.exc.IntegrityError:
-                LOG.exception('Problem finding entity to delete')
-                raise exception.NotFound("Entity ID %s not found"
-                                         % entity_id)
+        try:
+            entity.delete(session=session)
+        except sqlalchemy.exc.IntegrityError:
+            LOG.exception('Problem finding entity to delete')
+            raise exception.NotFound("Entity ID %s not found"
+                                     % entity_id)
 
     def _do_entity_name(self):
         """Sub-class hook: return entity name, such as for debugging."""
@@ -462,39 +498,38 @@ class BaseRepo(object):
                           find and update it
         """
         session = get_session()
-        with session.begin():
 
+        if entity_id:
+            entity_ref = self.get(entity_id, session=session)
+            values['updated_at'] = timeutils.utcnow()
+        else:
+            self._do_convert_values(values)
+            entity_ref = self._do_create_instance()
+
+        # Need to canonicalize ownership
+        if 'owner' in values and not values['owner']:
+            values['owner'] = None
+
+        entity_ref.update(values)
+
+        # Validate the attributes before we go any further. From my
+        # (unknown Glance developer) investigation, the @validates
+        # decorator does not validate
+        # on new records, only on existing records, which is, well,
+        # idiotic.
+        self._do_validate(entity_ref.to_dict())
+        self._update_values(entity_ref, values)
+
+        try:
+            entity_ref.save(session=session)
+        except sqlalchemy.exc.IntegrityError:
+            LOG.exception('Problem saving entity for _update')
             if entity_id:
-                entity_ref = self.get(entity_id, session=session)
-                values['updated_at'] = timeutils.utcnow()
+                raise exception.NotFound("Entity ID %s not found"
+                                         % entity_id)
             else:
-                self._do_convert_values(values)
-                entity_ref = self._do_create_instance()
-
-            # Need to canonicalize ownership
-            if 'owner' in values and not values['owner']:
-                values['owner'] = None
-
-            entity_ref.update(values)
-
-            # Validate the attributes before we go any further. From my
-            # (unknown Glance developer) investigation, the @validates
-            # decorator does not validate
-            # on new records, only on existing records, which is, well,
-            # idiotic.
-            self._do_validate(entity_ref.to_dict())
-            self._update_values(entity_ref, values)
-
-            try:
-                entity_ref.save(session=session)
-            except sqlalchemy.exc.IntegrityError:
-                LOG.exception('Problem saving entity for _update')
-                if entity_id:
-                    raise exception.NotFound("Entity ID %s not found"
-                                             % entity_id)
-                else:
-                    raise exception.Duplicate("Entity ID %s already exists!"
-                                              % values['id'])
+                raise exception.Duplicate("Entity ID %s already exists!"
+                                          % values['id'])
 
         return self.get(entity_ref.id)
 
@@ -664,30 +699,29 @@ class SecretStoreMetadatumRepo(BaseRepo):
         :raises NotFound if entity does not exist.
         """
         now = timeutils.utcnow()
-        session = get_session()
-        with session.begin():
-            for k, v in metadata.items():
-                meta_model = models.SecretStoreMetadatum(k, v)
-                meta_model.updated_at = now
-                meta_model.secret = secret_model
-                meta_model.save(session=session)
+
+        for k, v in metadata.items():
+            meta_model = models.SecretStoreMetadatum(k, v)
+            meta_model.updated_at = now
+            meta_model.secret = secret_model
+            meta_model.save()
 
     def get_metadata_for_secret(self, secret_id):
         """Returns a dict of SecretStoreMetadatum instances."""
 
         session = get_session()
-        with session.begin():
-            try:
-                query = session.query(models.SecretStoreMetadatum)
-                query = query.filter_by(deleted=False)
 
-                query = query.filter(
-                    models.SecretStoreMetadatum.secret_id == secret_id)
+        try:
+            query = session.query(models.SecretStoreMetadatum)
+            query = query.filter_by(deleted=False)
 
-                metadata = query.all()
+            query = query.filter(
+                models.SecretStoreMetadatum.secret_id == secret_id)
 
-            except sa_orm.exc.NoResultFound:
-                metadata = dict()
+            metadata = query.all()
+
+        except sa_orm.exc.NoResultFound:
+            metadata = dict()
 
         return dict((m.key, m.value) for m in metadata)
 
@@ -871,29 +905,29 @@ class OrderPluginMetadatumRepo(BaseRepo):
         """
         now = timeutils.utcnow()
         session = get_session()
-        with session.begin():
-            for k, v in metadata.items():
-                meta_model = models.OrderPluginMetadatum(k, v)
-                meta_model.updated_at = now
-                meta_model.order = order_model
-                meta_model.save(session=session)
+
+        for k, v in metadata.items():
+            meta_model = models.OrderPluginMetadatum(k, v)
+            meta_model.updated_at = now
+            meta_model.order = order_model
+            meta_model.save(session=session)
 
     def get_metadata_for_order(self, order_id):
         """Returns a dict of OrderPluginMetadatum instances."""
 
         session = get_session()
-        with session.begin():
-            try:
-                query = session.query(models.OrderPluginMetadatum)
-                query = query.filter_by(deleted=False)
 
-                query = query.filter(
-                    models.OrderPluginMetadatum.order_id == order_id)
+        try:
+            query = session.query(models.OrderPluginMetadatum)
+            query = query.filter_by(deleted=False)
 
-                metadata = query.all()
+            query = query.filter(
+                models.OrderPluginMetadatum.order_id == order_id)
 
-            except sa_orm.exc.NoResultFound:
-                metadata = dict()
+            metadata = query.all()
+
+        except sa_orm.exc.NoResultFound:
+            metadata = dict()
 
         return dict((m.key, m.value) for m in metadata)
 
@@ -1064,11 +1098,12 @@ class ContainerConsumerRepo(BaseRepo):
     def create_from(self, new_consumer, container):
         session = get_session()
         try:
-            with session.begin():
-                container.updated_at = timeutils.utcnow()
-                container.consumers.append(new_consumer)
-                container.save(session=session)
+            container.updated_at = timeutils.utcnow()
+            container.consumers.append(new_consumer)
+            container.save(session=session)
         except sqlalchemy.exc.IntegrityError:
+            session.rollback()  # We know consumer already exists.
+
             # This operation is idempotent, so log this and move on
             LOG.debug("Consumer %s already exists for container %s,"
                       " continuing...", (new_consumer.name, new_consumer.URL),

barbican/plugin/store_crypto.py

@@ -296,6 +296,7 @@ def _find_or_create_kek_objects(plugin_inst, tenant_model):
 
 def _store_secret_and_datum(
         context, secret_model, kek_datum_model, generated_dto):
+
     # Create Secret entities in data store.
     if not secret_model.id:
         repositories.get_secret_repository().create_from(secret_model)

barbican/queue/__init__.py

@@ -49,6 +49,7 @@ CONF.register_opts(queue_opts, group=queue_opt_group)
 
 
 TRANSPORT = None
+IS_SERVER_SIDE = True
 
 ALLOWED_EXMODS = [
     exception.__name__,
@@ -59,13 +60,18 @@ def get_allowed_exmods():
     return ALLOWED_EXMODS
 
 
-def init(conf):
-    global TRANSPORT
+def init(conf, is_server_side=True):
+    global TRANSPORT, IS_SERVER_SIDE
     exmods = get_allowed_exmods()
+    IS_SERVER_SIDE = is_server_side
     TRANSPORT = messaging.get_transport(conf,
                                         allowed_remote_exmods=exmods)
 
 
+def is_server_side():
+    return IS_SERVER_SIDE
+
+
 def cleanup():
     global TRANSPORT
     assert TRANSPORT is not None

barbican/queue/server.py

@@ -16,9 +16,12 @@
 """
 Server-side (i.e. worker side) classes and logic.
 """
+import functools
+
 from oslo.config import cfg
 
 from barbican.common import utils
+from barbican.model import repositories
 from barbican.openstack.common import service
 from barbican import queue
 from barbican.tasks import resources
@@ -29,6 +32,35 @@ LOG = utils.getLogger(__name__)
 CONF = cfg.CONF
 
 
+def transactional(fn):
+    """Provides request-scoped database transaction support to tasks."""
+
+    @functools.wraps(fn)
+    def wrapper(*args, **kwargs):
+        if not queue.is_server_side():
+            fn(*args, **kwargs)  # Non-server mode directly invokes tasks.
+        else:
+            # Start the database session.
+            repositories.start()
+
+            # Manage session/transaction.
+            try:
+                fn(*args, **kwargs)
+                repositories.commit()
+            except Exception:
+                """NOTE: Wrapped functions must process with care!
+
+                Exceptions that reach here will revert the entire transaction,
+                including any updates made to entities such as setting error
+                codes and error messages.
+                """
+                repositories.rollback()
+            finally:
+                repositories.clear()
+
+    return wrapper
+
+
 class Tasks(object):
     """Tasks that can be invoked asynchronously in Barbican.
 
@@ -41,6 +73,7 @@ class Tasks(object):
     methods on itself, which include the methods in this class.
     """
 
+    @transactional
     def process_type_order(self, context, order_id, keystone_id):
         """Process TypeOrder."""
         LOG.debug('TypeOrder id is {0}'.format(order_id))
@@ -51,6 +84,7 @@ class Tasks(object):
             LOG.exception(">>>>> Task exception seen, details reported "
                           "on the Orders entity.")
 
+    @transactional
     def update_order(self, context, order_id, keystone_id, updated_meta):
         """Update Order."""
         task = resources.UpdateOrder()

barbican/tests/queue/test_server.py

@@ -19,20 +19,129 @@ from barbican.queue import server
 from barbican.tests import utils
 
 
+class WhenUsingTransactionalDecorator(utils.BaseTestCase):
+    """Test using the 'transactional' decorator in server.py.
+
+    Note that only the 'I am a server' logic is tested here, as the alternate
+    mode is only used for direct invocation of Task methods in the standalone
+    server mode, which is also thoroughly tested in WhenUsingBeginTypeOrderTask
+    below.
+    """
+
+    def setUp(self):
+        super(WhenUsingTransactionalDecorator, self).setUp()
+
+        # Ensure we always thing we are in 'I am a server' mode.
+        is_server_side_config = {
+            'return_value': True
+        }
+        self.is_server_side_patcher = mock.patch(
+            'barbican.queue.is_server_side',
+            **is_server_side_config
+        )
+        self.is_server_side_patcher.start()
+
+        # Mock the various repository calls:
+        self.start_patcher = mock.patch(
+            'barbican.model.repositories.start'
+        )
+        self.start_mock = self.start_patcher.start()
+
+        self.commit_patcher = mock.patch(
+            'barbican.model.repositories.commit'
+        )
+        self.commit_mock = self.commit_patcher.start()
+
+        self.rollback_patcher = mock.patch(
+            'barbican.model.repositories.rollback'
+        )
+        self.rollback_mock = self.rollback_patcher.start()
+
+        self.clear_patcher = mock.patch(
+            'barbican.model.repositories.clear'
+        )
+        self.clear_mock = self.clear_patcher.start()
+
+        self.args = ('foo', 'bar')
+        self.kwargs = {'k_foo': 1, 'k_bar': 2}
+
+        # Class/decorator under test.
+        class TestClass(object):
+            my_args = None
+            my_kwargs = None
+            is_exception_needed = False
+
+            @server.transactional
+            def test_method(self, *args, **kwargs):
+                if self.is_exception_needed:
+                    raise ValueError()
+                self.my_args = args
+                self.my_kwargs = kwargs
+        self.test_object = TestClass()
+
+    def tearDown(self):
+        super(WhenUsingTransactionalDecorator, self).tearDown()
+        self.is_server_side_patcher.stop()
+        self.start_patcher.stop()
+        self.commit_patcher.stop()
+        self.rollback_patcher.stop()
+        self.clear_patcher.stop()
+
+    def test_should_commit(self):
+        self.test_object.test_method(*self.args, **self.kwargs)
+
+        self.assertEqual(self.args, self.test_object.my_args)
+        self.assertEqual(self.kwargs, self.test_object.my_kwargs)
+
+        self.assertEqual(self.start_mock.call_count, 1)
+        self.assertEqual(self.commit_mock.call_count, 1)
+        self.assertEqual(self.rollback_mock.call_count, 0)
+        self.assertEqual(self.clear_mock.call_count, 1)
+
+    def test_should_rollback(self):
+        self.test_object.is_exception_needed = True
+
+        self.test_object.test_method(*self.args, **self.kwargs)
+
+        self.assertEqual(self.start_mock.call_count, 1)
+        self.assertEqual(self.commit_mock.call_count, 0)
+        self.assertEqual(self.rollback_mock.call_count, 1)
+        self.assertEqual(self.clear_mock.call_count, 1)
+
+
 class WhenUsingBeginTypeOrderTask(utils.BaseTestCase):
     """Test using the Tasks class for 'type order' task."""
 
     def setUp(self):
         super(WhenUsingBeginTypeOrderTask, self).setUp()
 
+        # Mock the 'am I a server process?' flag used by the decorator around
+        #   all task methods. Since this test class focuses on testing task
+        #   method behaviors, this flag is set to false to allow for direct
+        #   testing of these tasks without database transactional interference.
+        is_server_side_config = {
+            'return_value': False
+        }
+        self.is_server_side_patcher = mock.patch(
+            'barbican.queue.is_server_side',
+            **is_server_side_config
+        )
+        self.is_server_side_patcher.start()
+
         self.tasks = server.Tasks()
 
+    def tearDown(self):
+        super(WhenUsingBeginTypeOrderTask, self).tearDown()
+        self.is_server_side_patcher.stop()
+
     @mock.patch('barbican.tasks.resources.BeginTypeOrder')
     def test_should_process_order(self, mock_begin_order):
         mock_begin_order.return_value.process.return_value = None
+
         self.tasks.process_type_order(context=None,
                                       order_id=self.order_id,
                                       keystone_id=self.keystone_id)
+
         mock_begin_order.return_value.process.assert_called_with(
             self.order_id, self.keystone_id)
 
@@ -40,6 +149,7 @@ class WhenUsingBeginTypeOrderTask(utils.BaseTestCase):
     def test_should_update_order(self, mock_update_order):
         mock_update_order.return_value.process.return_value = None
         updated_meta = {}
+
         self.tasks.update_order(context=None,
                                 order_id=self.order_id,
                                 keystone_id=self.keystone_id,
@@ -52,6 +162,7 @@ class WhenUsingBeginTypeOrderTask(utils.BaseTestCase):
     def test_process_order_catch_exception(self, mock_begin_order):
         """Test process_type_order() handles all exceptions."""
         mock_begin_order.return_value.process.side_effect = Exception()
+
         self.tasks.process_type_order(None, self.order_id, self.keystone_id)