Sync SQLAlchemy lifecycle to request cycle

SQLAlchemy session/transaction lifecycles should be synchronized to
request lifecycles, initiating with session binding to the database
engine, followed by data model processing, followed by either a
commit() or rollback(), and finally a session disposal once the request
is completed. This CR creates repository module functions for each of
these lifecycle operations. For the API process, this means utilizing
Pecan's transaction hook to invoke the SQLAlchemy lifecycle functions
above. For the Worker process, this means decorating each RPC method
invoked from queue to call the above lifecycle functions.

Change-Id: Id7216eb6871dae849f782a0f196210cd85d08f2f
Closes-Bug: #1376239

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)