diff --git a/barbican/api/app.py b/barbican/api/app.py index 5bf58c831..92c8d468c 100644 --- a/barbican/api/app.py +++ b/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 diff --git a/barbican/model/repositories.py b/barbican/model/repositories.py index c94fd7380..2e6b3cdee 100644 --- a/barbican/model/repositories.py +++ b/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), diff --git a/barbican/plugin/store_crypto.py b/barbican/plugin/store_crypto.py index 2ec84058b..7e2905241 100644 --- a/barbican/plugin/store_crypto.py +++ b/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) diff --git a/barbican/queue/__init__.py b/barbican/queue/__init__.py index 95666b438..cad5eed04 100644 --- a/barbican/queue/__init__.py +++ b/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 diff --git a/barbican/queue/server.py b/barbican/queue/server.py index 65109530e..b26f7fa97 100644 --- a/barbican/queue/server.py +++ b/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() diff --git a/barbican/tests/queue/test_server.py b/barbican/tests/queue/test_server.py index 1bbaf0c04..e9e5da8ac 100644 --- a/barbican/tests/queue/test_server.py +++ b/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)