From 0e69ae82b7ae97065dafd57f01ff1d3734a5f66b Mon Sep 17 00:00:00 2001 From: Andrew Lazarev Date: Thu, 29 May 2014 13:58:37 -0700 Subject: [PATCH] Sync the latest DB code from oslo-incubator Update includes a lot of changes from oslo. Previous update was on Dec 9, 2013. Sahara code changed because of oslo.db API change. The main change in API is that oslo.db no longer stores SQLAlchemy Engine and sessionmaker instances globally and it's up to applications to create them. Change-Id: I3e90739ab96fa8c4c718f22542df8460cb052ad9 --- etc/sahara/sahara.conf.sample | 54 +- etc/sahara/sahara.conf.sample-basic | 5 +- requirements.txt | 1 + sahara/db/api.py | 5 +- sahara/db/base.py | 3 +- sahara/db/sqlalchemy/api.py | 87 ++- sahara/openstack/common/context.py | 111 +++ sahara/openstack/common/db/api.py | 155 +++- sahara/openstack/common/db/exception.py | 6 +- sahara/openstack/common/db/options.py | 171 +++++ .../common/db/sqlalchemy/migration.py | 53 +- .../openstack/common/db/sqlalchemy/models.py | 18 +- .../common/db/sqlalchemy/provision.py | 88 +-- .../openstack/common/db/sqlalchemy/session.py | 689 +++++++++--------- .../common/db/sqlalchemy/test_base.py | 165 +++++ .../common/db/sqlalchemy/test_migrations.py | 18 +- .../openstack/common/db/sqlalchemy/utils.py | 133 +++- sahara/tests/unit/base.py | 4 +- .../unit/db/migration/test_migrations_base.py | 8 +- test-requirements.txt | 1 + 20 files changed, 1257 insertions(+), 518 deletions(-) create mode 100644 sahara/openstack/common/context.py create mode 100644 sahara/openstack/common/db/options.py create mode 100644 sahara/openstack/common/db/sqlalchemy/test_base.py diff --git a/etc/sahara/sahara.conf.sample b/etc/sahara/sahara.conf.sample index cfcda7c2f9..4cfc00b4da 100644 --- a/etc/sahara/sahara.conf.sample +++ b/etc/sahara/sahara.conf.sample @@ -86,17 +86,6 @@ #db_driver=sahara.db -# -# Options defined in sahara.openstack.common.db.sqlalchemy.session -# - -# The file name to use with SQLite (string value) -#sqlite_db=sahara.sqlite - -# If True, SQLite uses synchronous mode (boolean value) -#sqlite_synchronous=true - - # # Options defined in sahara.openstack.common.lockutils # @@ -316,28 +305,32 @@ [database] # -# Options defined in sahara.openstack.common.db.api +# Options defined in sahara.openstack.common.db.options # +# The file name to use with SQLite (string value) +#sqlite_db=sahara.sqlite + +# If True, SQLite uses synchronous mode (boolean value) +#sqlite_synchronous=true + # The backend to use for db (string value) # Deprecated group/name - [DEFAULT]/db_backend #backend=sqlalchemy - -# -# Options defined in sahara.openstack.common.db.sqlalchemy.session -# - # The SQLAlchemy connection string used to connect to the # database (string value) # Deprecated group/name - [DEFAULT]/sql_connection # Deprecated group/name - [DATABASE]/sql_connection # Deprecated group/name - [sql]/connection -#connection=sqlite:////sahara/openstack/common/db/$sqlite_db +#connection= -# The SQLAlchemy connection string used to connect to the -# slave database (string value) -#slave_connection= +# The SQL mode to be used for MySQL sessions. This option, +# including the default, overrides any server-set SQL mode. To +# use whatever SQL mode is set by the server configuration, +# set this to no value. Example: mysql_sql_mode= (string +# value) +#mysql_sql_mode=TRADITIONAL # Timeout before idle sql connections are reaped (integer # value) @@ -391,4 +384,23 @@ # Deprecated group/name - [DATABASE]/sqlalchemy_pool_timeout #pool_timeout= +# Enable the experimental use of database reconnect on +# connection lost (boolean value) +#use_db_reconnect=false + +# seconds between db connection retries (integer value) +#db_retry_interval=1 + +# Whether to increase interval between db connection retries, +# up to db_max_retry_interval (boolean value) +#db_inc_retry_interval=true + +# max seconds between db connection retries, if +# db_inc_retry_interval is enabled (integer value) +#db_max_retry_interval=10 + +# maximum db connection retries before error is raised. +# (setting -1 implies an infinite retry count) (integer value) +#db_max_retries=20 + diff --git a/etc/sahara/sahara.conf.sample-basic b/etc/sahara/sahara.conf.sample-basic index 7284386c83..db50235688 100644 --- a/etc/sahara/sahara.conf.sample-basic +++ b/etc/sahara/sahara.conf.sample-basic @@ -107,5 +107,8 @@ #plugins=vanilla,hdp [database] -#connection=sqlite:////sahara/openstack/common/db/$sqlite_db + +# The SQLAlchemy connection string used to connect to the +# database (string value) +#connection= diff --git a/requirements.txt b/requirements.txt index 5ab6809bcc..160307024f 100644 --- a/requirements.txt +++ b/requirements.txt @@ -4,6 +4,7 @@ eventlet>=0.13.0 Flask>=0.10,<1.0 iso8601>=0.1.9 jsonschema>=2.0.0,<3.0.0 +lockfile>=0.8 oslo.config>=1.2.0 oslo.messaging>=1.3.0 paramiko>=1.13.0 diff --git a/sahara/db/api.py b/sahara/db/api.py index f2a3818a67..beb5fd361c 100644 --- a/sahara/db/api.py +++ b/sahara/db/api.py @@ -39,11 +39,14 @@ from sahara.openstack.common import log as logging CONF = cfg.CONF +CONF.import_opt('backend', 'sahara.openstack.common.db.options', + group='database') + _BACKEND_MAPPING = { 'sqlalchemy': 'sahara.db.sqlalchemy.api', } -IMPL = db_api.DBAPI(backend_mapping=_BACKEND_MAPPING) +IMPL = db_api.DBAPI(CONF.database.backend, backend_mapping=_BACKEND_MAPPING) LOG = logging.getLogger(__name__) diff --git a/sahara/db/base.py b/sahara/db/base.py index fcf430726d..bdbdf1c9a7 100644 --- a/sahara/db/base.py +++ b/sahara/db/base.py @@ -36,4 +36,5 @@ class Base(object): def is_mysql_avail(): - return CONF.database.connection.startswith('mysql') + connection = CONF.database.connection + return connection and connection.startswith('mysql') diff --git a/sahara/db/sqlalchemy/api.py b/sahara/db/sqlalchemy/api.py index 7120e08123..80ad00d797 100644 --- a/sahara/db/sqlalchemy/api.py +++ b/sahara/db/sqlalchemy/api.py @@ -30,11 +30,38 @@ from sahara.openstack.common import log as logging LOG = logging.getLogger(__name__) -get_engine = db_session.get_engine -get_session = db_session.get_session - CONF = cfg.CONF +_FACADE = None + + +def _create_facade_lazily(): + global _FACADE + + if _FACADE is None: + params = dict(CONF.database.iteritems()) + params["sqlite_fk"] = True + _FACADE = db_session.EngineFacade( + CONF.database.connection, + **params + ) + return _FACADE + + +def get_engine(): + facade = _create_facade_lazily() + return facade.get_engine() + + +def get_session(**kwargs): + facade = _create_facade_lazily() + return facade.get_session(**kwargs) + + +def cleanup(): + global _FACADE + _FACADE = None + def get_backend(): """The backend is this module itself.""" @@ -72,7 +99,7 @@ def count_query(model, context, session=None, project_only=None): def setup_db(): try: - engine = db_session.get_engine(sqlite_fk=True) + engine = get_engine() m.Cluster.metadata.create_all(engine) except sa.exc.OperationalError as e: LOG.exception("Database registration exception: %s", e) @@ -82,7 +109,7 @@ def setup_db(): def drop_db(): try: - engine = db_session.get_engine(sqlite_fk=True) + engine = get_engine() m.Cluster.metadata.drop_all(engine) except Exception as e: LOG.exception("Database shutdown exception: %s", e) @@ -400,11 +427,13 @@ def node_group_template_create(context, values): node_group_template = m.NodeGroupTemplate() node_group_template.update(values) - try: - node_group_template.save() - except db_exc.DBDuplicateEntry as e: - raise ex.DBDuplicateEntry("Duplicate entry for NodeGroupTemplate: %s" - % e.columns) + session = get_session() + with session.begin(): + try: + node_group_template.save(session=session) + except db_exc.DBDuplicateEntry as e: + raise ex.DBDuplicateEntry("Duplicate entry for NodeGroupTemplate: " + "%s" % e.columns) return node_group_template @@ -442,11 +471,13 @@ def data_source_create(context, values): data_source = m.DataSource() data_source.update(values) - try: - data_source.save() - except db_exc.DBDuplicateEntry as e: - raise ex.DBDuplicateEntry("Duplicate entry for DataSource: %s" - % e.columns) + session = get_session() + with session.begin(): + try: + data_source.save(session=session) + except db_exc.DBDuplicateEntry as e: + raise ex.DBDuplicateEntry("Duplicate entry for DataSource: %s" + % e.columns) return data_source @@ -494,7 +525,7 @@ def job_execution_create(context, values): job_ex = m.JobExecution() job_ex.update(values) try: - job_ex.save() + job_ex.save(session=session) except db_exc.DBDuplicateEntry as e: raise ex.DBDuplicateEntry("Duplicate entry for JobExecution: %s" % e.columns) @@ -635,11 +666,13 @@ def job_binary_create(context, values): job_binary = m.JobBinary() job_binary.update(values) - try: - job_binary.save() - except db_exc.DBDuplicateEntry as e: - raise ex.DBDuplicateEntry("Duplicate entry for JobBinary: %s" - % e.columns) + session = get_session() + with session.begin(): + try: + job_binary.save(session=session) + except db_exc.DBDuplicateEntry as e: + raise ex.DBDuplicateEntry("Duplicate entry for JobBinary: %s" + % e.columns) return job_binary @@ -728,11 +761,13 @@ def job_binary_internal_create(context, values): job_binary_int = m.JobBinaryInternal() job_binary_int.update(values) - try: - job_binary_int.save() - except db_exc.DBDuplicateEntry as e: - raise ex.DBDuplicateEntry("Duplicate entry for JobBinaryInternal: %s" - % e.columns) + session = get_session() + with session.begin(): + try: + job_binary_int.save(session=session) + except db_exc.DBDuplicateEntry as e: + raise ex.DBDuplicateEntry("Duplicate entry for JobBinaryInternal: " + "%s" % e.columns) return job_binary_internal_get(context, job_binary_int.id) diff --git a/sahara/openstack/common/context.py b/sahara/openstack/common/context.py new file mode 100644 index 0000000000..3eeb445e48 --- /dev/null +++ b/sahara/openstack/common/context.py @@ -0,0 +1,111 @@ +# Copyright 2011 OpenStack Foundation. +# 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. + +""" +Simple class that stores security context information in the web request. + +Projects should subclass this class if they wish to enhance the request +context or provide additional information in their specific WSGI pipeline. +""" + +import itertools +import uuid + + +def generate_request_id(): + return b'req-' + str(uuid.uuid4()).encode('ascii') + + +class RequestContext(object): + + """Helper class to represent useful information about a request context. + + Stores information about the security context under which the user + accesses the system, as well as additional request information. + """ + + user_idt_format = '{user} {tenant} {domain} {user_domain} {p_domain}' + + def __init__(self, auth_token=None, user=None, tenant=None, domain=None, + user_domain=None, project_domain=None, is_admin=False, + read_only=False, show_deleted=False, request_id=None, + instance_uuid=None): + self.auth_token = auth_token + self.user = user + self.tenant = tenant + self.domain = domain + self.user_domain = user_domain + self.project_domain = project_domain + self.is_admin = is_admin + self.read_only = read_only + self.show_deleted = show_deleted + self.instance_uuid = instance_uuid + if not request_id: + request_id = generate_request_id() + self.request_id = request_id + + def to_dict(self): + user_idt = ( + self.user_idt_format.format(user=self.user or '-', + tenant=self.tenant or '-', + domain=self.domain or '-', + user_domain=self.user_domain or '-', + p_domain=self.project_domain or '-')) + + return {'user': self.user, + 'tenant': self.tenant, + 'domain': self.domain, + 'user_domain': self.user_domain, + 'project_domain': self.project_domain, + 'is_admin': self.is_admin, + 'read_only': self.read_only, + 'show_deleted': self.show_deleted, + 'auth_token': self.auth_token, + 'request_id': self.request_id, + 'instance_uuid': self.instance_uuid, + 'user_identity': user_idt} + + +def get_admin_context(show_deleted=False): + context = RequestContext(None, + tenant=None, + is_admin=True, + show_deleted=show_deleted) + return context + + +def get_context_from_function_and_args(function, args, kwargs): + """Find an arg of type RequestContext and return it. + + This is useful in a couple of decorators where we don't + know much about the function we're wrapping. + """ + + for arg in itertools.chain(kwargs.values(), args): + if isinstance(arg, RequestContext): + return arg + + return None + + +def is_user_context(context): + """Indicates if the request context is a normal user.""" + if not context: + return False + if context.is_admin: + return False + if not context.user_id or not context.project_id: + return False + return True diff --git a/sahara/openstack/common/db/api.py b/sahara/openstack/common/db/api.py index 0411a5c71c..b44fd5bcc8 100644 --- a/sahara/openstack/common/db/api.py +++ b/sahara/openstack/common/db/api.py @@ -15,43 +15,148 @@ """Multiple DB API backend support. -Supported configuration options: - -The following two parameters are in the 'database' group: -`backend`: DB backend name or full module path to DB backend module. - A DB backend module should implement a method named 'get_backend' which takes no arguments. The method can return any object that implements DB API methods. """ -from oslo.config import cfg +import functools +import logging +import threading +import time +from sahara.openstack.common.db import exception +from sahara.openstack.common.gettextutils import _LE from sahara.openstack.common import importutils -db_opts = [ - cfg.StrOpt('backend', - default='sqlalchemy', - deprecated_name='db_backend', - deprecated_group='DEFAULT', - help='The backend to use for db'), -] +LOG = logging.getLogger(__name__) -CONF = cfg.CONF -CONF.register_opts(db_opts, 'database') + +def safe_for_db_retry(f): + """Enable db-retry for decorated function, if config option enabled.""" + f.__dict__['enable_retry'] = True + return f + + +class wrap_db_retry(object): + """Retry db.api methods, if DBConnectionError() raised + + Retry decorated db.api methods. If we enabled `use_db_reconnect` + in config, this decorator will be applied to all db.api functions, + marked with @safe_for_db_retry decorator. + Decorator catchs DBConnectionError() and retries function in a + loop until it succeeds, or until maximum retries count will be reached. + """ + + def __init__(self, retry_interval, max_retries, inc_retry_interval, + max_retry_interval): + super(wrap_db_retry, self).__init__() + + self.retry_interval = retry_interval + self.max_retries = max_retries + self.inc_retry_interval = inc_retry_interval + self.max_retry_interval = max_retry_interval + + def __call__(self, f): + @functools.wraps(f) + def wrapper(*args, **kwargs): + next_interval = self.retry_interval + remaining = self.max_retries + + while True: + try: + return f(*args, **kwargs) + except exception.DBConnectionError as e: + if remaining == 0: + LOG.exception(_LE('DB exceeded retry limit.')) + raise exception.DBError(e) + if remaining != -1: + remaining -= 1 + LOG.exception(_LE('DB connection error.')) + # NOTE(vsergeyev): We are using patched time module, so + # this effectively yields the execution + # context to another green thread. + time.sleep(next_interval) + if self.inc_retry_interval: + next_interval = min( + next_interval * 2, + self.max_retry_interval + ) + return wrapper class DBAPI(object): - def __init__(self, backend_mapping=None): - if backend_mapping is None: - backend_mapping = {} - backend_name = CONF.database.backend - # Import the untranslated name if we don't have a - # mapping. - backend_path = backend_mapping.get(backend_name, backend_name) - backend_mod = importutils.import_module(backend_path) - self.__backend = backend_mod.get_backend() + def __init__(self, backend_name, backend_mapping=None, lazy=False, + **kwargs): + """Initialize the chosen DB API backend. + + :param backend_name: name of the backend to load + :type backend_name: str + + :param backend_mapping: backend name -> module/class to load mapping + :type backend_mapping: dict + + :param lazy: load the DB backend lazily on the first DB API method call + :type lazy: bool + + Keyword arguments: + + :keyword use_db_reconnect: retry DB transactions on disconnect or not + :type use_db_reconnect: bool + + :keyword retry_interval: seconds between transaction retries + :type retry_interval: int + + :keyword inc_retry_interval: increase retry interval or not + :type inc_retry_interval: bool + + :keyword max_retry_interval: max interval value between retries + :type max_retry_interval: int + + :keyword max_retries: max number of retries before an error is raised + :type max_retries: int + + """ + + self._backend = None + self._backend_name = backend_name + self._backend_mapping = backend_mapping or {} + self._lock = threading.Lock() + + if not lazy: + self._load_backend() + + self.use_db_reconnect = kwargs.get('use_db_reconnect', False) + self.retry_interval = kwargs.get('retry_interval', 1) + self.inc_retry_interval = kwargs.get('inc_retry_interval', True) + self.max_retry_interval = kwargs.get('max_retry_interval', 10) + self.max_retries = kwargs.get('max_retries', 20) + + def _load_backend(self): + with self._lock: + if not self._backend: + # Import the untranslated name if we don't have a mapping + backend_path = self._backend_mapping.get(self._backend_name, + self._backend_name) + backend_mod = importutils.import_module(backend_path) + self._backend = backend_mod.get_backend() def __getattr__(self, key): - return getattr(self.__backend, key) + if not self._backend: + self._load_backend() + + attr = getattr(self._backend, key) + if not hasattr(attr, '__call__'): + return attr + # NOTE(vsergeyev): If `use_db_reconnect` option is set to True, retry + # DB API methods, decorated with @safe_for_db_retry + # on disconnect. + if self.use_db_reconnect and hasattr(attr, 'enable_retry'): + attr = wrap_db_retry( + retry_interval=self.retry_interval, + max_retries=self.max_retries, + inc_retry_interval=self.inc_retry_interval, + max_retry_interval=self.max_retry_interval)(attr) + + return attr diff --git a/sahara/openstack/common/db/exception.py b/sahara/openstack/common/db/exception.py index 9374ffe3ee..f94b82e1fd 100644 --- a/sahara/openstack/common/db/exception.py +++ b/sahara/openstack/common/db/exception.py @@ -16,6 +16,8 @@ """DB related custom exceptions.""" +import six + from sahara.openstack.common.gettextutils import _ @@ -23,7 +25,7 @@ class DBError(Exception): """Wraps an implementation specific exception.""" def __init__(self, inner_exception=None): self.inner_exception = inner_exception - super(DBError, self).__init__(str(inner_exception)) + super(DBError, self).__init__(six.text_type(inner_exception)) class DBDuplicateEntry(DBError): @@ -46,7 +48,7 @@ class DBInvalidUnicodeParameter(Exception): class DbMigrationError(DBError): """Wraps migration specific exception.""" def __init__(self, message=None): - super(DbMigrationError, self).__init__(str(message)) + super(DbMigrationError, self).__init__(message) class DBConnectionError(DBError): diff --git a/sahara/openstack/common/db/options.py b/sahara/openstack/common/db/options.py new file mode 100644 index 0000000000..28b4e20537 --- /dev/null +++ b/sahara/openstack/common/db/options.py @@ -0,0 +1,171 @@ +# 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. + +import copy + +from oslo.config import cfg + + +database_opts = [ + cfg.StrOpt('sqlite_db', + deprecated_group='DEFAULT', + default='sahara.sqlite', + help='The file name to use with SQLite'), + cfg.BoolOpt('sqlite_synchronous', + deprecated_group='DEFAULT', + default=True, + help='If True, SQLite uses synchronous mode'), + cfg.StrOpt('backend', + default='sqlalchemy', + deprecated_name='db_backend', + deprecated_group='DEFAULT', + help='The backend to use for db'), + cfg.StrOpt('connection', + help='The SQLAlchemy connection string used to connect to the ' + 'database', + secret=True, + deprecated_opts=[cfg.DeprecatedOpt('sql_connection', + group='DEFAULT'), + cfg.DeprecatedOpt('sql_connection', + group='DATABASE'), + cfg.DeprecatedOpt('connection', + group='sql'), ]), + cfg.StrOpt('mysql_sql_mode', + default='TRADITIONAL', + help='The SQL mode to be used for MySQL sessions. ' + 'This option, including the default, overrides any ' + 'server-set SQL mode. To use whatever SQL mode ' + 'is set by the server configuration, ' + 'set this to no value. Example: mysql_sql_mode='), + cfg.IntOpt('idle_timeout', + default=3600, + deprecated_opts=[cfg.DeprecatedOpt('sql_idle_timeout', + group='DEFAULT'), + cfg.DeprecatedOpt('sql_idle_timeout', + group='DATABASE'), + cfg.DeprecatedOpt('idle_timeout', + group='sql')], + help='Timeout before idle sql connections are reaped'), + cfg.IntOpt('min_pool_size', + default=1, + deprecated_opts=[cfg.DeprecatedOpt('sql_min_pool_size', + group='DEFAULT'), + cfg.DeprecatedOpt('sql_min_pool_size', + group='DATABASE')], + help='Minimum number of SQL connections to keep open in a ' + 'pool'), + cfg.IntOpt('max_pool_size', + default=None, + deprecated_opts=[cfg.DeprecatedOpt('sql_max_pool_size', + group='DEFAULT'), + cfg.DeprecatedOpt('sql_max_pool_size', + group='DATABASE')], + help='Maximum number of SQL connections to keep open in a ' + 'pool'), + cfg.IntOpt('max_retries', + default=10, + deprecated_opts=[cfg.DeprecatedOpt('sql_max_retries', + group='DEFAULT'), + cfg.DeprecatedOpt('sql_max_retries', + group='DATABASE')], + help='Maximum db connection retries during startup. ' + '(setting -1 implies an infinite retry count)'), + cfg.IntOpt('retry_interval', + default=10, + deprecated_opts=[cfg.DeprecatedOpt('sql_retry_interval', + group='DEFAULT'), + cfg.DeprecatedOpt('reconnect_interval', + group='DATABASE')], + help='Interval between retries of opening a sql connection'), + cfg.IntOpt('max_overflow', + default=None, + deprecated_opts=[cfg.DeprecatedOpt('sql_max_overflow', + group='DEFAULT'), + cfg.DeprecatedOpt('sqlalchemy_max_overflow', + group='DATABASE')], + help='If set, use this value for max_overflow with sqlalchemy'), + cfg.IntOpt('connection_debug', + default=0, + deprecated_opts=[cfg.DeprecatedOpt('sql_connection_debug', + group='DEFAULT')], + help='Verbosity of SQL debugging information. 0=None, ' + '100=Everything'), + cfg.BoolOpt('connection_trace', + default=False, + deprecated_opts=[cfg.DeprecatedOpt('sql_connection_trace', + group='DEFAULT')], + help='Add python stack traces to SQL as comment strings'), + cfg.IntOpt('pool_timeout', + default=None, + deprecated_opts=[cfg.DeprecatedOpt('sqlalchemy_pool_timeout', + group='DATABASE')], + help='If set, use this value for pool_timeout with sqlalchemy'), + cfg.BoolOpt('use_db_reconnect', + default=False, + help='Enable the experimental use of database reconnect ' + 'on connection lost'), + cfg.IntOpt('db_retry_interval', + default=1, + help='seconds between db connection retries'), + cfg.BoolOpt('db_inc_retry_interval', + default=True, + help='Whether to increase interval between db connection ' + 'retries, up to db_max_retry_interval'), + cfg.IntOpt('db_max_retry_interval', + default=10, + help='max seconds between db connection retries, if ' + 'db_inc_retry_interval is enabled'), + cfg.IntOpt('db_max_retries', + default=20, + help='maximum db connection retries before error is raised. ' + '(setting -1 implies an infinite retry count)'), +] + +CONF = cfg.CONF +CONF.register_opts(database_opts, 'database') + + +def set_defaults(sql_connection, sqlite_db, max_pool_size=None, + max_overflow=None, pool_timeout=None): + """Set defaults for configuration variables.""" + cfg.set_defaults(database_opts, + connection=sql_connection, + sqlite_db=sqlite_db) + # Update the QueuePool defaults + if max_pool_size is not None: + cfg.set_defaults(database_opts, + max_pool_size=max_pool_size) + if max_overflow is not None: + cfg.set_defaults(database_opts, + max_overflow=max_overflow) + if pool_timeout is not None: + cfg.set_defaults(database_opts, + pool_timeout=pool_timeout) + + +def list_opts(): + """Returns a list of oslo.config options available in the library. + + The returned list includes all oslo.config options which may be registered + at runtime by the library. + + Each element of the list is a tuple. The first element is the name of the + group under which the list of elements in the second element will be + registered. A group name of None corresponds to the [DEFAULT] group in + config files. + + The purpose of this is to allow tools like the Oslo sample config file + generator to discover the options exposed to users by this library. + + :returns: a list of (group_name, opts) tuples + """ + return [('database', copy.deepcopy(database_opts))] diff --git a/sahara/openstack/common/db/sqlalchemy/migration.py b/sahara/openstack/common/db/sqlalchemy/migration.py index b0b99f6d34..a247686900 100644 --- a/sahara/openstack/common/db/sqlalchemy/migration.py +++ b/sahara/openstack/common/db/sqlalchemy/migration.py @@ -51,13 +51,9 @@ import sqlalchemy from sqlalchemy.schema import UniqueConstraint from sahara.openstack.common.db import exception -from sahara.openstack.common.db.sqlalchemy import session as db_session from sahara.openstack.common.gettextutils import _ -get_engine = db_session.get_engine - - def _get_unique_constraints(self, table): """Retrieve information about existing unique constraints of the table @@ -172,17 +168,20 @@ def patch_migrate(): sqlite.SQLiteConstraintGenerator) -def db_sync(abs_path, version=None, init_version=0): +def db_sync(engine, abs_path, version=None, init_version=0, sanity_check=True): """Upgrade or downgrade a database. Function runs the upgrade() or downgrade() functions in change scripts. + :param engine: SQLAlchemy engine instance for a given database :param abs_path: Absolute path to migrate repository. :param version: Database will upgrade/downgrade until this version. If None - database will update to the latest available version. :param init_version: Initial database version + :param sanity_check: Require schema sanity checking for all tables """ + if version is not None: try: version = int(version) @@ -190,49 +189,62 @@ def db_sync(abs_path, version=None, init_version=0): raise exception.DbMigrationError( message=_("version should be an integer")) - current_version = db_version(abs_path, init_version) + current_version = db_version(engine, abs_path, init_version) repository = _find_migrate_repo(abs_path) - _db_schema_sanity_check() + if sanity_check: + _db_schema_sanity_check(engine) if version is None or version > current_version: - return versioning_api.upgrade(get_engine(), repository, version) + return versioning_api.upgrade(engine, repository, version) else: - return versioning_api.downgrade(get_engine(), repository, + return versioning_api.downgrade(engine, repository, version) -def _db_schema_sanity_check(): - engine = get_engine() +def _db_schema_sanity_check(engine): + """Ensure all database tables were created with required parameters. + + :param engine: SQLAlchemy engine instance for a given database + + """ + if engine.name == 'mysql': onlyutf8_sql = ('SELECT TABLE_NAME,TABLE_COLLATION ' 'from information_schema.TABLES ' 'where TABLE_SCHEMA=%s and ' 'TABLE_COLLATION NOT LIKE "%%utf8%%"') - table_names = [res[0] for res in engine.execute(onlyutf8_sql, - engine.url.database)] + # NOTE(morganfainberg): exclude the sqlalchemy-migrate and alembic + # versioning tables from the tables we need to verify utf8 status on. + # Non-standard table names are not supported. + EXCLUDED_TABLES = ['migrate_version', 'alembic_version'] + + table_names = [res[0] for res in + engine.execute(onlyutf8_sql, engine.url.database) if + res[0].lower() not in EXCLUDED_TABLES] + if len(table_names) > 0: raise ValueError(_('Tables "%s" have non utf8 collation, ' 'please make sure all tables are CHARSET=utf8' ) % ','.join(table_names)) -def db_version(abs_path, init_version): +def db_version(engine, abs_path, init_version): """Show the current version of the repository. + :param engine: SQLAlchemy engine instance for a given database :param abs_path: Absolute path to migrate repository :param version: Initial database version """ repository = _find_migrate_repo(abs_path) try: - return versioning_api.db_version(get_engine(), repository) + return versioning_api.db_version(engine, repository) except versioning_exceptions.DatabaseNotControlledError: meta = sqlalchemy.MetaData() - engine = get_engine() meta.reflect(bind=engine) tables = meta.tables if len(tables) == 0 or 'alembic_version' in tables: - db_version_control(abs_path, init_version) - return versioning_api.db_version(get_engine(), repository) + db_version_control(engine, abs_path, version=init_version) + return versioning_api.db_version(engine, repository) else: raise exception.DbMigrationError( message=_( @@ -241,17 +253,18 @@ def db_version(abs_path, init_version): "manually.")) -def db_version_control(abs_path, version=None): +def db_version_control(engine, abs_path, version=None): """Mark a database as under this repository's version control. Once a database is under version control, schema changes should only be done via change scripts in this repository. + :param engine: SQLAlchemy engine instance for a given database :param abs_path: Absolute path to migrate repository :param version: Initial database version """ repository = _find_migrate_repo(abs_path) - versioning_api.version_control(get_engine(), repository, version) + versioning_api.version_control(engine, repository, version) return version diff --git a/sahara/openstack/common/db/sqlalchemy/models.py b/sahara/openstack/common/db/sqlalchemy/models.py index 4f3f856a42..fe4567baf7 100644 --- a/sahara/openstack/common/db/sqlalchemy/models.py +++ b/sahara/openstack/common/db/sqlalchemy/models.py @@ -26,18 +26,16 @@ from sqlalchemy import Column, Integer from sqlalchemy import DateTime from sqlalchemy.orm import object_mapper -from sahara.openstack.common.db.sqlalchemy import session as sa from sahara.openstack.common import timeutils -class ModelBase(object): +class ModelBase(six.Iterator): """Base class for models.""" __table_initialized__ = False - def save(self, session=None): + def save(self, session): """Save this object.""" - if not session: - session = sa.get_session() + # NOTE(boris-42): This part of code should be look like: # session.add(self) # session.flush() @@ -72,7 +70,7 @@ class ModelBase(object): return [] def __iter__(self): - columns = dict(object_mapper(self).columns).keys() + columns = list(dict(object_mapper(self).columns).keys()) # NOTE(russellb): Allow models to specify other keys that can be looked # up, beyond the actual db columns. An example would be the 'name' # property for an Instance. @@ -80,10 +78,14 @@ class ModelBase(object): self._i = iter(columns) return self - def next(self): + # In Python 3, __next__() has replaced next(). + def __next__(self): n = six.advance_iterator(self._i) return n, getattr(self, n) + def next(self): + return self.__next__() + def update(self, values): """Make the model object behave like a dict.""" for k, v in six.iteritems(values): @@ -110,7 +112,7 @@ class SoftDeleteMixin(object): deleted_at = Column(DateTime) deleted = Column(Integer, default=0) - def soft_delete(self, session=None): + def soft_delete(self, session): """Mark this object as deleted.""" self.deleted = self.id self.deleted_at = timeutils.utcnow() diff --git a/sahara/openstack/common/db/sqlalchemy/provision.py b/sahara/openstack/common/db/sqlalchemy/provision.py index 1b4fa03a9d..798b091226 100644 --- a/sahara/openstack/common/db/sqlalchemy/provision.py +++ b/sahara/openstack/common/db/sqlalchemy/provision.py @@ -16,6 +16,7 @@ """Provision test environment for specific DB backends""" import argparse +import logging import os import random import string @@ -26,23 +27,12 @@ import sqlalchemy from sahara.openstack.common.db import exception as exc -SQL_CONNECTION = os.getenv('OS_TEST_DBAPI_ADMIN_CONNECTION', 'sqlite://') +LOG = logging.getLogger(__name__) -def _gen_credentials(*names): - """Generate credentials.""" - auth_dict = {} - for name in names: - val = ''.join(random.choice(string.ascii_lowercase) - for i in moves.range(10)) - auth_dict[name] = val - return auth_dict - - -def _get_engine(uri=SQL_CONNECTION): +def get_engine(uri): """Engine creation - By default the uri is SQL_CONNECTION which is admin credentials. Call the function without arguments to get admin connection. Admin connection required to create temporary user and database for each particular test. Otherwise use existing connection to recreate connection @@ -62,50 +52,43 @@ def _execute_sql(engine, sql, driver): except sqlalchemy.exc.OperationalError: msg = ('%s does not match database admin ' 'credentials or database does not exist.') - raise exc.DBConnectionError(msg % SQL_CONNECTION) + LOG.exception(msg % engine.url) + raise exc.DBConnectionError(msg % engine.url) def create_database(engine): """Provide temporary user and database for each particular test.""" driver = engine.name - auth = _gen_credentials('database', 'user', 'passwd') - - sqls = { - 'mysql': [ - "drop database if exists %(database)s;", - "grant all on %(database)s.* to '%(user)s'@'localhost'" - " identified by '%(passwd)s';", - "create database %(database)s;", - ], - 'postgresql': [ - "drop database if exists %(database)s;", - "drop user if exists %(user)s;", - "create user %(user)s with password '%(passwd)s';", - "create database %(database)s owner %(user)s;", - ] + auth = { + 'database': ''.join(random.choice(string.ascii_lowercase) + for i in moves.range(10)), + 'user': engine.url.username, + 'passwd': engine.url.password, } + sqls = [ + "drop database if exists %(database)s;", + "create database %(database)s;" + ] + if driver == 'sqlite': return 'sqlite:////tmp/%s' % auth['database'] - - try: - sql_rows = sqls[driver] - except KeyError: + elif driver in ['mysql', 'postgresql']: + sql_query = map(lambda x: x % auth, sqls) + _execute_sql(engine, sql_query, driver) + else: raise ValueError('Unsupported RDBMS %s' % driver) - sql_query = map(lambda x: x % auth, sql_rows) - - _execute_sql(engine, sql_query, driver) params = auth.copy() params['backend'] = driver return "%(backend)s://%(user)s:%(passwd)s@localhost/%(database)s" % params -def drop_database(engine, current_uri): +def drop_database(admin_engine, current_uri): """Drop temporary database and user after each particular test.""" - engine = _get_engine(current_uri) - admin_engine = _get_engine() + + engine = get_engine(current_uri) driver = engine.name auth = {'database': engine.url.database, 'user': engine.url.username} @@ -114,26 +97,11 @@ def drop_database(engine, current_uri): os.remove(auth['database']) except OSError: pass - return - - sqls = { - 'mysql': [ - "drop database if exists %(database)s;", - "drop user '%(user)s'@'localhost';", - ], - 'postgresql': [ - "drop database if exists %(database)s;", - "drop user if exists %(user)s;", - ] - } - - try: - sql_rows = sqls[driver] - except KeyError: + elif driver in ['mysql', 'postgresql']: + sql = "drop database if exists %(database)s;" + _execute_sql(admin_engine, [sql % auth], driver) + else: raise ValueError('Unsupported RDBMS %s' % driver) - sql_query = map(lambda x: x % auth, sql_rows) - - _execute_sql(admin_engine, sql_query, driver) def main(): @@ -172,7 +140,9 @@ def main(): args = parser.parse_args() - engine = _get_engine() + connection_string = os.getenv('OS_TEST_DBAPI_ADMIN_CONNECTION', + 'sqlite://') + engine = get_engine(connection_string) which = args.which if which == "create": diff --git a/sahara/openstack/common/db/sqlalchemy/session.py b/sahara/openstack/common/db/sqlalchemy/session.py index 603d9cb682..9f29b37e13 100644 --- a/sahara/openstack/common/db/sqlalchemy/session.py +++ b/sahara/openstack/common/db/sqlalchemy/session.py @@ -16,33 +16,24 @@ """Session Handling for SQLAlchemy backend. -Initializing: - -* Call set_defaults with the minimal of the following kwargs: - sql_connection, sqlite_db - - Example:: - - session.set_defaults( - sql_connection="sqlite:///var/lib/sahara/sqlite.db", - sqlite_db="/var/lib/sahara/sqlite.db") - Recommended ways to use sessions within this framework: -* Don't use them explicitly; this is like running with AUTOCOMMIT=1. - model_query() will implicitly use a session when called without one +* Don't use them explicitly; this is like running with ``AUTOCOMMIT=1``. + `model_query()` will implicitly use a session when called without one supplied. This is the ideal situation because it will allow queries to be automatically retried if the database connection is interrupted. - Note: Automatic retry will be enabled in a future patch. + .. note:: Automatic retry will be enabled in a future patch. It is generally fine to issue several queries in a row like this. Even though they may be run in separate transactions and/or separate sessions, each one will see the data from the prior calls. If needed, undo- or rollback-like functionality should be handled at a logical level. For an example, look at - the code around quotas and reservation_rollback(). + the code around quotas and `reservation_rollback()`. - Examples:: + Examples: + + .. code:: python def get_foo(context, foo): return (model_query(context, models.Foo). @@ -61,28 +52,29 @@ Recommended ways to use sessions within this framework: return foo_ref -* Within the scope of a single method, keeping all the reads and writes within - the context managed by a single session. In this way, the session's __exit__ - handler will take care of calling flush() and commit() for you. - If using this approach, you should not explicitly call flush() or commit(). - Any error within the context of the session will cause the session to emit - a ROLLBACK. Database Errors like IntegrityError will be raised in - session's __exit__ handler, and any try/except within the context managed - by session will not be triggered. And catching other non-database errors in - the session will not trigger the ROLLBACK, so exception handlers should - always be outside the session, unless the developer wants to do a partial - commit on purpose. If the connection is dropped before this is possible, - the database will implicitly roll back the transaction. +* Within the scope of a single method, keep all the reads and writes within + the context managed by a single session. In this way, the session's + `__exit__` handler will take care of calling `flush()` and `commit()` for + you. If using this approach, you should not explicitly call `flush()` or + `commit()`. Any error within the context of the session will cause the + session to emit a `ROLLBACK`. Database errors like `IntegrityError` will be + raised in `session`'s `__exit__` handler, and any try/except within the + context managed by `session` will not be triggered. And catching other + non-database errors in the session will not trigger the ROLLBACK, so + exception handlers should always be outside the session, unless the + developer wants to do a partial commit on purpose. If the connection is + dropped before this is possible, the database will implicitly roll back the + transaction. - Note: statements in the session scope will not be automatically retried. + .. note:: Statements in the session scope will not be automatically retried. If you create models within the session, they need to be added, but you - do not need to call model.save() + do not need to call `model.save()`: - :: + .. code:: python def create_many_foo(context, foos): - session = get_session() + session = sessionmaker() with session.begin(): for foo in foos: foo_ref = models.Foo() @@ -90,7 +82,7 @@ Recommended ways to use sessions within this framework: session.add(foo_ref) def update_bar(context, foo_id, newbar): - session = get_session() + session = sessionmaker() with session.begin(): foo_ref = (model_query(context, models.Foo, session). filter_by(id=foo_id). @@ -99,11 +91,16 @@ Recommended ways to use sessions within this framework: filter_by(id=foo_ref['bar_id']). update({'bar': newbar})) - Note: update_bar is a trivially simple example of using "with session.begin". - Whereas create_many_foo is a good example of when a transaction is needed, - it is always best to use as few queries as possible. The two queries in - update_bar can be better expressed using a single query which avoids - the need for an explicit transaction. It can be expressed like so:: + .. note:: `update_bar` is a trivially simple example of using + ``with session.begin``. Whereas `create_many_foo` is a good example of + when a transaction is needed, it is always best to use as few queries as + possible. + + The two queries in `update_bar` can be better expressed using a single query + which avoids the need for an explicit transaction. It can be expressed like + so: + + .. code:: python def update_bar(context, foo_id, newbar): subq = (model_query(context, models.Foo.id). @@ -114,21 +111,25 @@ Recommended ways to use sessions within this framework: filter_by(id=subq.as_scalar()). update({'bar': newbar})) - For reference, this emits approximately the following SQL statement:: + For reference, this emits approximately the following SQL statement: + + .. code:: sql UPDATE bar SET bar = ${newbar} WHERE id=(SELECT bar_id FROM foo WHERE id = ${foo_id} LIMIT 1); - Note: create_duplicate_foo is a trivially simple example of catching an - exception while using "with session.begin". Here create two duplicate - instances with same primary key, must catch the exception out of context - managed by a single session: + .. note:: `create_duplicate_foo` is a trivially simple example of catching an + exception while using ``with session.begin``. Here create two duplicate + instances with same primary key, must catch the exception out of context + managed by a single session: + + .. code:: python def create_duplicate_foo(context): foo1 = models.Foo() foo2 = models.Foo() foo1.id = foo2.id = 1 - session = get_session() + session = sessionmaker() try: with session.begin(): session.add(foo1) @@ -138,7 +139,7 @@ Recommended ways to use sessions within this framework: * Passing an active session between methods. Sessions should only be passed to private methods. The private method must use a subtransaction; otherwise - SQLAlchemy will throw an error when you call session.begin() on an existing + SQLAlchemy will throw an error when you call `session.begin()` on an existing transaction. Public methods should not accept a session parameter and should not be involved in sessions within the caller's scope. @@ -151,10 +152,10 @@ Recommended ways to use sessions within this framework: becomes less clear in this situation. When this is needed for code clarity, it should be clearly documented. - :: + .. code:: python def myfunc(foo): - session = get_session() + session = sessionmaker() with session.begin(): # do some database things bar = _private_func(foo, session) @@ -162,7 +163,7 @@ Recommended ways to use sessions within this framework: def _private_func(foo, session=None): if not session: - session = get_session() + session = sessionmaker() with session.begin(subtransaction=True): # do some other database things return bar @@ -172,13 +173,13 @@ There are some things which it is best to avoid: * Don't keep a transaction open any longer than necessary. - This means that your "with session.begin()" block should be as short + This means that your ``with session.begin()`` block should be as short as possible, while still containing all the related calls for that transaction. -* Avoid "with_lockmode('UPDATE')" when possible. +* Avoid ``with_lockmode('UPDATE')`` when possible. - In MySQL/InnoDB, when a "SELECT ... FOR UPDATE" query does not match + In MySQL/InnoDB, when a ``SELECT ... FOR UPDATE`` query does not match any rows, it will take a gap-lock. This is a form of write-lock on the "gap" where no rows exist, and prevents any other writes to that space. This can effectively prevent any INSERT into a table by locking the gap @@ -189,15 +190,18 @@ There are some things which it is best to avoid: number of rows matching a query, and if only one row is returned, then issue the SELECT FOR UPDATE. - The better long-term solution is to use INSERT .. ON DUPLICATE KEY UPDATE. + The better long-term solution is to use + ``INSERT .. ON DUPLICATE KEY UPDATE``. However, this can not be done until the "deleted" columns are removed and proper UNIQUE constraints are added to the tables. Enabling soft deletes: -* To use/enable soft-deletes, the SoftDeleteMixin must be added - to your model class. For example:: +* To use/enable soft-deletes, the `SoftDeleteMixin` must be added + to your model class. For example: + + .. code:: python class NovaBase(models.SoftDeleteMixin, models.ModelBase): pass @@ -205,15 +209,16 @@ Enabling soft deletes: Efficient use of soft deletes: -* There are two possible ways to mark a record as deleted:: +* There are two possible ways to mark a record as deleted: + `model.soft_delete()` and `query.soft_delete()`. - model.soft_delete() and query.soft_delete(). + The `model.soft_delete()` method works with a single already-fetched entry. + `query.soft_delete()` makes only one db request for all entries that + correspond to the query. - model.soft_delete() method works with single already fetched entry. - query.soft_delete() makes only one db request for all entries that correspond - to query. +* In almost all cases you should use `query.soft_delete()`. Some examples: -* In almost all cases you should use query.soft_delete(). Some examples:: + .. code:: python def soft_delete_bar(): count = model_query(BarModel).find(some_condition).soft_delete() @@ -222,7 +227,7 @@ Efficient use of soft deletes: def complex_soft_delete_with_synchronization_bar(session=None): if session is None: - session = get_session() + session = sessionmaker() with session.begin(subtransactions=True): count = (model_query(BarModel). find(some_condition). @@ -232,24 +237,26 @@ Efficient use of soft deletes: if count == 0: raise Exception("0 entries were soft deleted") -* There is only one situation where model.soft_delete() is appropriate: when +* There is only one situation where `model.soft_delete()` is appropriate: when you fetch a single record, work with it, and mark it as deleted in the same transaction. - :: + .. code:: python def soft_delete_bar_model(): - session = get_session() + session = sessionmaker() with session.begin(): bar_ref = model_query(BarModel).find(some_condition).first() # Work with bar_ref bar_ref.soft_delete(session=session) However, if you need to work with all entries that correspond to query and - then soft delete them you should use query.soft_delete() method:: + then soft delete them you should use the `query.soft_delete()` method: + + .. code:: python def soft_delete_multi_models(): - session = get_session() + session = sessionmaker() with session.begin(): query = (model_query(BarModel, session=session). find(some_condition)) @@ -260,22 +267,22 @@ Efficient use of soft deletes: # session and these entries are not used after this. When working with many rows, it is very important to use query.soft_delete, - which issues a single query. Using model.soft_delete(), as in the following + which issues a single query. Using `model.soft_delete()`, as in the following example, is very inefficient. - :: + .. code:: python for bar_ref in bar_refs: bar_ref.soft_delete(session=session) # This will produce count(bar_refs) db requests. + """ import functools -import os.path +import logging import re import time -from oslo.config import cfg import six from sqlalchemy import exc as sqla_exc from sqlalchemy.interfaces import PoolListener @@ -284,151 +291,12 @@ from sqlalchemy.pool import NullPool, StaticPool from sqlalchemy.sql.expression import literal_column from sahara.openstack.common.db import exception -from sahara.openstack.common.gettextutils import _ -from sahara.openstack.common import log as logging +from sahara.openstack.common.gettextutils import _LE, _LW from sahara.openstack.common import timeutils -sqlite_db_opts = [ - cfg.StrOpt('sqlite_db', - default='sahara.sqlite', - help='The file name to use with SQLite'), - cfg.BoolOpt('sqlite_synchronous', - default=True, - help='If True, SQLite uses synchronous mode'), -] - -database_opts = [ - cfg.StrOpt('connection', - default='sqlite:///' + - os.path.abspath(os.path.join(os.path.dirname(__file__), - '../', '$sqlite_db')), - help='The SQLAlchemy connection string used to connect to the ' - 'database', - secret=True, - deprecated_opts=[cfg.DeprecatedOpt('sql_connection', - group='DEFAULT'), - cfg.DeprecatedOpt('sql_connection', - group='DATABASE'), - cfg.DeprecatedOpt('connection', - group='sql'), ]), - cfg.StrOpt('slave_connection', - default='', - secret=True, - help='The SQLAlchemy connection string used to connect to the ' - 'slave database'), - cfg.IntOpt('idle_timeout', - default=3600, - deprecated_opts=[cfg.DeprecatedOpt('sql_idle_timeout', - group='DEFAULT'), - cfg.DeprecatedOpt('sql_idle_timeout', - group='DATABASE'), - cfg.DeprecatedOpt('idle_timeout', - group='sql')], - help='Timeout before idle sql connections are reaped'), - cfg.IntOpt('min_pool_size', - default=1, - deprecated_opts=[cfg.DeprecatedOpt('sql_min_pool_size', - group='DEFAULT'), - cfg.DeprecatedOpt('sql_min_pool_size', - group='DATABASE')], - help='Minimum number of SQL connections to keep open in a ' - 'pool'), - cfg.IntOpt('max_pool_size', - default=None, - deprecated_opts=[cfg.DeprecatedOpt('sql_max_pool_size', - group='DEFAULT'), - cfg.DeprecatedOpt('sql_max_pool_size', - group='DATABASE')], - help='Maximum number of SQL connections to keep open in a ' - 'pool'), - cfg.IntOpt('max_retries', - default=10, - deprecated_opts=[cfg.DeprecatedOpt('sql_max_retries', - group='DEFAULT'), - cfg.DeprecatedOpt('sql_max_retries', - group='DATABASE')], - help='Maximum db connection retries during startup. ' - '(setting -1 implies an infinite retry count)'), - cfg.IntOpt('retry_interval', - default=10, - deprecated_opts=[cfg.DeprecatedOpt('sql_retry_interval', - group='DEFAULT'), - cfg.DeprecatedOpt('reconnect_interval', - group='DATABASE')], - help='Interval between retries of opening a sql connection'), - cfg.IntOpt('max_overflow', - default=None, - deprecated_opts=[cfg.DeprecatedOpt('sql_max_overflow', - group='DEFAULT'), - cfg.DeprecatedOpt('sqlalchemy_max_overflow', - group='DATABASE')], - help='If set, use this value for max_overflow with sqlalchemy'), - cfg.IntOpt('connection_debug', - default=0, - deprecated_opts=[cfg.DeprecatedOpt('sql_connection_debug', - group='DEFAULT')], - help='Verbosity of SQL debugging information. 0=None, ' - '100=Everything'), - cfg.BoolOpt('connection_trace', - default=False, - deprecated_opts=[cfg.DeprecatedOpt('sql_connection_trace', - group='DEFAULT')], - help='Add python stack traces to SQL as comment strings'), - cfg.IntOpt('pool_timeout', - default=None, - deprecated_opts=[cfg.DeprecatedOpt('sqlalchemy_pool_timeout', - group='DATABASE')], - help='If set, use this value for pool_timeout with sqlalchemy'), -] - -CONF = cfg.CONF -CONF.register_opts(sqlite_db_opts) -CONF.register_opts(database_opts, 'database') LOG = logging.getLogger(__name__) -_ENGINE = None -_MAKER = None -_SLAVE_ENGINE = None -_SLAVE_MAKER = None - - -def set_defaults(sql_connection, sqlite_db, max_pool_size=None, - max_overflow=None, pool_timeout=None): - """Set defaults for configuration variables.""" - cfg.set_defaults(database_opts, - connection=sql_connection) - cfg.set_defaults(sqlite_db_opts, - sqlite_db=sqlite_db) - # Update the QueuePool defaults - if max_pool_size is not None: - cfg.set_defaults(database_opts, - max_pool_size=max_pool_size) - if max_overflow is not None: - cfg.set_defaults(database_opts, - max_overflow=max_overflow) - if pool_timeout is not None: - cfg.set_defaults(database_opts, - pool_timeout=pool_timeout) - - -def cleanup(): - global _ENGINE, _MAKER - global _SLAVE_ENGINE, _SLAVE_MAKER - - if _MAKER: - _MAKER.close_all() - _MAKER = None - if _ENGINE: - _ENGINE.dispose() - _ENGINE = None - if _SLAVE_MAKER: - _SLAVE_MAKER.close_all() - _SLAVE_MAKER = None - if _SLAVE_ENGINE: - _SLAVE_ENGINE.dispose() - _SLAVE_ENGINE = None - class SqliteForeignKeysListener(PoolListener): """Ensures that the foreign key constraints are enforced in SQLite. @@ -441,30 +309,6 @@ class SqliteForeignKeysListener(PoolListener): dbapi_con.execute('pragma foreign_keys=ON') -def get_session(autocommit=True, expire_on_commit=False, sqlite_fk=False, - slave_session=False, mysql_traditional_mode=False): - """Return a SQLAlchemy session.""" - global _MAKER - global _SLAVE_MAKER - maker = _MAKER - - if slave_session: - maker = _SLAVE_MAKER - - if maker is None: - engine = get_engine(sqlite_fk=sqlite_fk, slave_engine=slave_session, - mysql_traditional_mode=mysql_traditional_mode) - maker = get_maker(engine, autocommit, expire_on_commit) - - if slave_session: - _SLAVE_MAKER = maker - else: - _MAKER = maker - - session = maker() - return session - - # note(boris-42): In current versions of DB backends unique constraint # violation messages follow the structure: # @@ -473,9 +317,9 @@ def get_session(autocommit=True, expire_on_commit=False, sqlite_fk=False, # N columns - (IntegrityError) column c1, c2, ..., N are not unique # # sqlite since 3.7.16: -# 1 column - (IntegrityError) UNIQUE constraint failed: k1 +# 1 column - (IntegrityError) UNIQUE constraint failed: tbl.k1 # -# N columns - (IntegrityError) UNIQUE constraint failed: k1, k2 +# N columns - (IntegrityError) UNIQUE constraint failed: tbl.k1, tbl.k2 # # postgres: # 1 column - (IntegrityError) duplicate key value violates unique @@ -488,11 +332,20 @@ def get_session(autocommit=True, expire_on_commit=False, sqlite_fk=False, # 'c1'") # N columns - (IntegrityError) (1062, "Duplicate entry 'values joined # with -' for key 'name_of_our_constraint'") +# +# ibm_db_sa: +# N columns - (IntegrityError) SQL0803N One or more values in the INSERT +# statement, UPDATE statement, or foreign key update caused by a +# DELETE statement are not valid because the primary key, unique +# constraint or unique index identified by "2" constrains table +# "NOVA.KEY_PAIRS" from having duplicate values for the index +# key. _DUP_KEY_RE_DB = { "sqlite": (re.compile(r"^.*columns?([^)]+)(is|are)\s+not\s+unique$"), re.compile(r"^.*UNIQUE\s+constraint\s+failed:\s+(.+)$")), "postgresql": (re.compile(r"^.*duplicate\s+key.*\"([^\"]+)\"\s*\n.*$"),), - "mysql": (re.compile(r"^.*\(1062,.*'([^\']+)'\"\)$"),) + "mysql": (re.compile(r"^.*\(1062,.*'([^\']+)'\"\)$"),), + "ibm_db_sa": (re.compile(r"^.*SQL0803N.*$"),), } @@ -514,7 +367,7 @@ def _raise_if_duplicate_entry_error(integrity_error, engine_name): return [columns] return columns[len(uniqbase):].split("0")[1:] - if engine_name not in ["mysql", "sqlite", "postgresql"]: + if engine_name not in ("ibm_db_sa", "mysql", "sqlite", "postgresql"): return # FIXME(johannes): The usage of the .message attribute has been @@ -529,10 +382,15 @@ def _raise_if_duplicate_entry_error(integrity_error, engine_name): else: return - columns = match.group(1) + # NOTE(mriedem): The ibm_db_sa integrity error message doesn't provide the + # columns so we have to omit that from the DBDuplicateEntry error. + columns = '' + + if engine_name != 'ibm_db_sa': + columns = match.group(1) if engine_name == "sqlite": - columns = columns.strip().split(", ") + columns = [c.split('.')[-1] for c in columns.strip().split(", ")] else: columns = get_columns_from_uniq_cons_or_name(columns) raise exception.DBDuplicateEntry(columns, integrity_error) @@ -571,56 +429,39 @@ def _raise_if_deadlock_error(operational_error, engine_name): def _wrap_db_error(f): @functools.wraps(f) - def _wrap(*args, **kwargs): + def _wrap(self, *args, **kwargs): try: - return f(*args, **kwargs) + assert issubclass( + self.__class__, sqlalchemy.orm.session.Session + ), ('_wrap_db_error() can only be applied to methods of ' + 'subclasses of sqlalchemy.orm.session.Session.') + + return f(self, *args, **kwargs) except UnicodeEncodeError: raise exception.DBInvalidUnicodeParameter() - # note(boris-42): We should catch unique constraint violation and - # wrap it by our own DBDuplicateEntry exception. Unique constraint - # violation is wrapped by IntegrityError. except sqla_exc.OperationalError as e: - _raise_if_deadlock_error(e, get_engine().name) + _raise_if_db_connection_lost(e, self.bind) + _raise_if_deadlock_error(e, self.bind.dialect.name) # NOTE(comstud): A lot of code is checking for OperationalError # so let's not wrap it for now. raise + # note(boris-42): We should catch unique constraint violation and + # wrap it by our own DBDuplicateEntry exception. Unique constraint + # violation is wrapped by IntegrityError. except sqla_exc.IntegrityError as e: # note(boris-42): SqlAlchemy doesn't unify errors from different # DBs so we must do this. Also in some tables (for example # instance_types) there are more than one unique constraint. This # means we should get names of columns, which values violate # unique constraint, from error message. - _raise_if_duplicate_entry_error(e, get_engine().name) + _raise_if_duplicate_entry_error(e, self.bind.dialect.name) raise exception.DBError(e) except Exception as e: - LOG.exception(_('DB exception wrapped.')) + LOG.exception(_LE('DB exception wrapped.')) raise exception.DBError(e) return _wrap -def get_engine(sqlite_fk=False, slave_engine=False, - mysql_traditional_mode=False): - """Return a SQLAlchemy engine.""" - global _ENGINE - global _SLAVE_ENGINE - engine = _ENGINE - db_uri = CONF.database.connection - - if slave_engine: - engine = _SLAVE_ENGINE - db_uri = CONF.database.slave_connection - - if engine is None: - engine = create_engine(db_uri, sqlite_fk=sqlite_fk, - mysql_traditional_mode=mysql_traditional_mode) - if slave_engine: - _SLAVE_ENGINE = engine - else: - _ENGINE = engine - - return engine - - def _synchronous_switch_listener(dbapi_conn, connection_rec): """Switch sqlite connections to non-synchronous mode.""" dbapi_conn.execute("PRAGMA synchronous = OFF") @@ -648,7 +489,7 @@ def _thread_yield(dbapi_con, con_record): def _ping_listener(engine, dbapi_conn, connection_rec, connection_proxy): - """Ensures that MySQL and DB2 connections are alive. + """Ensures that MySQL, PostgreSQL or DB2 connections are alive. Borrowed from: http://groups.google.com/group/sqlalchemy/msg/a4ce563d802c929f @@ -662,22 +503,78 @@ def _ping_listener(engine, dbapi_conn, connection_rec, connection_proxy): cursor.execute(ping_sql) except Exception as ex: if engine.dialect.is_disconnect(ex, dbapi_conn, cursor): - msg = _('Database server has gone away: %s') % ex + msg = _LW('Database server has gone away: %s') % ex LOG.warning(msg) + + # if the database server has gone away, all connections in the pool + # have become invalid and we can safely close all of them here, + # rather than waste time on checking of every single connection + engine.dispose() + + # this will be handled by SQLAlchemy and will force it to create + # a new connection and retry the original action raise sqla_exc.DisconnectionError(msg) else: raise -def _set_mode_traditional(dbapi_con, connection_rec, connection_proxy): - """Set engine mode to 'traditional'. +def _set_session_sql_mode(dbapi_con, connection_rec, sql_mode=None): + """Set the sql_mode session variable. - Required to prevent silent truncates at insert or update operations - under MySQL. By default MySQL truncates inserted string if it longer - than a declared field just with warning. That is fraught with data - corruption. + MySQL supports several server modes. The default is None, but sessions + may choose to enable server modes like TRADITIONAL, ANSI, + several STRICT_* modes and others. + + Note: passing in '' (empty string) for sql_mode clears + the SQL mode for the session, overriding a potentially set + server default. """ - dbapi_con.cursor().execute("SET SESSION sql_mode = TRADITIONAL;") + + cursor = dbapi_con.cursor() + cursor.execute("SET SESSION sql_mode = %s", [sql_mode]) + + +def _mysql_get_effective_sql_mode(engine): + """Returns the effective SQL mode for connections from the engine pool. + + Returns ``None`` if the mode isn't available, otherwise returns the mode. + + """ + # Get the real effective SQL mode. Even when unset by + # our own config, the server may still be operating in a specific + # SQL mode as set by the server configuration. + # Also note that the checkout listener will be called on execute to + # set the mode if it's registered. + row = engine.execute("SHOW VARIABLES LIKE 'sql_mode'").fetchone() + if row is None: + return + return row[1] + + +def _mysql_check_effective_sql_mode(engine): + """Logs a message based on the effective SQL mode for MySQL connections.""" + realmode = _mysql_get_effective_sql_mode(engine) + + if realmode is None: + LOG.warning(_LW('Unable to detect effective SQL mode')) + return + + LOG.debug('MySQL server mode set to %s', realmode) + # 'TRADITIONAL' mode enables several other modes, so + # we need a substring match here + if not ('TRADITIONAL' in realmode.upper() or + 'STRICT_ALL_TABLES' in realmode.upper()): + LOG.warning(_LW("MySQL SQL mode is '%s', " + "consider enabling TRADITIONAL or STRICT_ALL_TABLES"), + realmode) + + +def _mysql_set_mode_callback(engine, sql_mode): + if sql_mode is not None: + mode_callback = functools.partial(_set_session_sql_mode, + sql_mode=sql_mode) + sqlalchemy.event.listen(engine, 'connect', mode_callback) + _mysql_check_effective_sql_mode(engine) def _is_db_connection_error(args): @@ -692,66 +589,75 @@ def _is_db_connection_error(args): return False -def create_engine(sql_connection, sqlite_fk=False, - mysql_traditional_mode=False): +def _raise_if_db_connection_lost(error, engine): + # NOTE(vsergeyev): Function is_disconnect(e, connection, cursor) + # requires connection and cursor in incoming parameters, + # but we have no possibility to create connection if DB + # is not available, so in such case reconnect fails. + # But is_disconnect() ignores these parameters, so it + # makes sense to pass to function None as placeholder + # instead of connection and cursor. + if engine.dialect.is_disconnect(error, None, None): + raise exception.DBConnectionError(error) + + +def create_engine(sql_connection, sqlite_fk=False, mysql_sql_mode=None, + idle_timeout=3600, + connection_debug=0, max_pool_size=None, max_overflow=None, + pool_timeout=None, sqlite_synchronous=True, + connection_trace=False, max_retries=10, retry_interval=10): """Return a new SQLAlchemy engine.""" - # NOTE(geekinutah): At this point we could be connecting to the normal - # db handle or the slave db handle. Things like - # _wrap_db_error aren't going to work well if their - # backends don't match. Let's check. - _assert_matching_drivers() + connection_dict = sqlalchemy.engine.url.make_url(sql_connection) engine_args = { - "pool_recycle": CONF.database.idle_timeout, - "echo": False, + "pool_recycle": idle_timeout, 'convert_unicode': True, } - # Map our SQL debug level to SQLAlchemy's options - if CONF.database.connection_debug >= 100: - engine_args['echo'] = 'debug' - elif CONF.database.connection_debug >= 50: - engine_args['echo'] = True + logger = logging.getLogger('sqlalchemy.engine') + + # Map SQL debug level to Python log level + if connection_debug >= 100: + logger.setLevel(logging.DEBUG) + elif connection_debug >= 50: + logger.setLevel(logging.INFO) + else: + logger.setLevel(logging.WARNING) if "sqlite" in connection_dict.drivername: if sqlite_fk: engine_args["listeners"] = [SqliteForeignKeysListener()] engine_args["poolclass"] = NullPool - if CONF.database.connection == "sqlite://": + if sql_connection == "sqlite://": engine_args["poolclass"] = StaticPool engine_args["connect_args"] = {'check_same_thread': False} else: - if CONF.database.max_pool_size is not None: - engine_args['pool_size'] = CONF.database.max_pool_size - if CONF.database.max_overflow is not None: - engine_args['max_overflow'] = CONF.database.max_overflow - if CONF.database.pool_timeout is not None: - engine_args['pool_timeout'] = CONF.database.pool_timeout + if max_pool_size is not None: + engine_args['pool_size'] = max_pool_size + if max_overflow is not None: + engine_args['max_overflow'] = max_overflow + if pool_timeout is not None: + engine_args['pool_timeout'] = pool_timeout engine = sqlalchemy.create_engine(sql_connection, **engine_args) sqlalchemy.event.listen(engine, 'checkin', _thread_yield) - if engine.name in ['mysql', 'ibm_db_sa']: - callback = functools.partial(_ping_listener, engine) - sqlalchemy.event.listen(engine, 'checkout', callback) - if mysql_traditional_mode: - sqlalchemy.event.listen(engine, 'checkout', _set_mode_traditional) - else: - LOG.warning(_("This application has not enabled MySQL traditional" - " mode, which means silent data corruption may" - " occur. Please encourage the application" - " developers to enable this mode.")) + if engine.name in ('ibm_db_sa', 'mysql', 'postgresql'): + ping_callback = functools.partial(_ping_listener, engine) + sqlalchemy.event.listen(engine, 'checkout', ping_callback) + if engine.name == 'mysql': + if mysql_sql_mode: + _mysql_set_mode_callback(engine, mysql_sql_mode) elif 'sqlite' in connection_dict.drivername: - if not CONF.sqlite_synchronous: + if not sqlite_synchronous: sqlalchemy.event.listen(engine, 'connect', _synchronous_switch_listener) sqlalchemy.event.listen(engine, 'connect', _add_regexp_listener) - if (CONF.database.connection_trace and - engine.dialect.dbapi.__name__ == 'MySQLdb'): + if connection_trace and engine.dialect.dbapi.__name__ == 'MySQLdb': _patch_mysqldb_with_stacktrace_comments() try: @@ -760,15 +666,15 @@ def create_engine(sql_connection, sqlite_fk=False, if not _is_db_connection_error(e.args[0]): raise - remaining = CONF.database.max_retries + remaining = max_retries if remaining == -1: remaining = 'infinite' while True: - msg = _('SQL connection failed. %s attempts left.') + msg = _LW('SQL connection failed. %s attempts left.') LOG.warning(msg % remaining) if remaining != 'infinite': remaining -= 1 - time.sleep(CONF.database.retry_interval) + time.sleep(retry_interval) try: engine.connect() break @@ -855,13 +761,144 @@ def _patch_mysqldb_with_stacktrace_comments(): setattr(MySQLdb.cursors.BaseCursor, '_do_query', _do_query) -def _assert_matching_drivers(): - """Make sure slave handle and normal handle have the same driver.""" - # NOTE(geekinutah): There's no use case for writing to one backend and - # reading from another. Who knows what the future holds? - if CONF.database.slave_connection == '': - return +class EngineFacade(object): + """A helper class for removing of global engine instances from sahara.db. - normal = sqlalchemy.engine.url.make_url(CONF.database.connection) - slave = sqlalchemy.engine.url.make_url(CONF.database.slave_connection) - assert normal.drivername == slave.drivername + As a library, sahara.db can't decide where to store/when to create engine + and sessionmaker instances, so this must be left for a target application. + + On the other hand, in order to simplify the adoption of sahara.db changes, + we'll provide a helper class, which creates engine and sessionmaker + on its instantiation and provides get_engine()/get_session() methods + that are compatible with corresponding utility functions that currently + exist in target projects, e.g. in Nova. + + engine/sessionmaker instances will still be global (and they are meant to + be global), but they will be stored in the app context, rather that in the + sahara.db context. + + Note: using of this helper is completely optional and you are encouraged to + integrate engine/sessionmaker instances into your apps any way you like + (e.g. one might want to bind a session to a request context). Two important + things to remember: + + 1. An Engine instance is effectively a pool of DB connections, so it's + meant to be shared (and it's thread-safe). + 2. A Session instance is not meant to be shared and represents a DB + transactional context (i.e. it's not thread-safe). sessionmaker is + a factory of sessions. + + """ + + def __init__(self, sql_connection, + sqlite_fk=False, autocommit=True, + expire_on_commit=False, **kwargs): + """Initialize engine and sessionmaker instances. + + :param sqlite_fk: enable foreign keys in SQLite + :type sqlite_fk: bool + + :param autocommit: use autocommit mode for created Session instances + :type autocommit: bool + + :param expire_on_commit: expire session objects on commit + :type expire_on_commit: bool + + Keyword arguments: + + :keyword mysql_sql_mode: the SQL mode to be used for MySQL sessions. + (defaults to TRADITIONAL) + :keyword idle_timeout: timeout before idle sql connections are reaped + (defaults to 3600) + :keyword connection_debug: verbosity of SQL debugging information. + 0=None, 100=Everything (defaults to 0) + :keyword max_pool_size: maximum number of SQL connections to keep open + in a pool (defaults to SQLAlchemy settings) + :keyword max_overflow: if set, use this value for max_overflow with + sqlalchemy (defaults to SQLAlchemy settings) + :keyword pool_timeout: if set, use this value for pool_timeout with + sqlalchemy (defaults to SQLAlchemy settings) + :keyword sqlite_synchronous: if True, SQLite uses synchronous mode + (defaults to True) + :keyword connection_trace: add python stack traces to SQL as comment + strings (defaults to False) + :keyword max_retries: maximum db connection retries during startup. + (setting -1 implies an infinite retry count) + (defaults to 10) + :keyword retry_interval: interval between retries of opening a sql + connection (defaults to 10) + + """ + + super(EngineFacade, self).__init__() + + self._engine = create_engine( + sql_connection=sql_connection, + sqlite_fk=sqlite_fk, + mysql_sql_mode=kwargs.get('mysql_sql_mode', 'TRADITIONAL'), + idle_timeout=kwargs.get('idle_timeout', 3600), + connection_debug=kwargs.get('connection_debug', 0), + max_pool_size=kwargs.get('max_pool_size'), + max_overflow=kwargs.get('max_overflow'), + pool_timeout=kwargs.get('pool_timeout'), + sqlite_synchronous=kwargs.get('sqlite_synchronous', True), + connection_trace=kwargs.get('connection_trace', False), + max_retries=kwargs.get('max_retries', 10), + retry_interval=kwargs.get('retry_interval', 10)) + self._session_maker = get_maker( + engine=self._engine, + autocommit=autocommit, + expire_on_commit=expire_on_commit) + + def get_engine(self): + """Get the engine instance (note, that it's shared).""" + + return self._engine + + def get_session(self, **kwargs): + """Get a Session instance. + + If passed, keyword arguments values override the ones used when the + sessionmaker instance was created. + + :keyword autocommit: use autocommit mode for created Session instances + :type autocommit: bool + + :keyword expire_on_commit: expire session objects on commit + :type expire_on_commit: bool + + """ + + for arg in kwargs: + if arg not in ('autocommit', 'expire_on_commit'): + del kwargs[arg] + + return self._session_maker(**kwargs) + + @classmethod + def from_config(cls, connection_string, conf, + sqlite_fk=False, autocommit=True, expire_on_commit=False): + """Initialize EngineFacade using oslo.config config instance options. + + :param connection_string: SQLAlchemy connection string + :type connection_string: string + + :param conf: oslo.config config instance + :type conf: oslo.config.cfg.ConfigOpts + + :param sqlite_fk: enable foreign keys in SQLite + :type sqlite_fk: bool + + :param autocommit: use autocommit mode for created Session instances + :type autocommit: bool + + :param expire_on_commit: expire session objects on commit + :type expire_on_commit: bool + + """ + + return cls(sql_connection=connection_string, + sqlite_fk=sqlite_fk, + autocommit=autocommit, + expire_on_commit=expire_on_commit, + **dict(conf.database.items())) diff --git a/sahara/openstack/common/db/sqlalchemy/test_base.py b/sahara/openstack/common/db/sqlalchemy/test_base.py new file mode 100644 index 0000000000..d4351df4c6 --- /dev/null +++ b/sahara/openstack/common/db/sqlalchemy/test_base.py @@ -0,0 +1,165 @@ +# Copyright (c) 2013 OpenStack Foundation +# 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. + +import abc +import functools +import os + +import fixtures +from oslotest import base as test_base +import six + +from sahara.openstack.common.db.sqlalchemy import provision +from sahara.openstack.common.db.sqlalchemy import session +from sahara.openstack.common.db.sqlalchemy import utils + + +class DbFixture(fixtures.Fixture): + """Basic database fixture. + + Allows to run tests on various db backends, such as SQLite, MySQL and + PostgreSQL. By default use sqlite backend. To override default backend + uri set env variable OS_TEST_DBAPI_CONNECTION with database admin + credentials for specific backend. + """ + + def _get_uri(self): + return os.getenv('OS_TEST_DBAPI_CONNECTION', 'sqlite://') + + def __init__(self, test): + super(DbFixture, self).__init__() + + self.test = test + + def cleanUp(self): + self.test.engine.dispose() + + def setUp(self): + super(DbFixture, self).setUp() + + self.test.engine = session.create_engine(self._get_uri()) + self.test.sessionmaker = session.get_maker(self.test.engine) + + +class DbTestCase(test_base.BaseTestCase): + """Base class for testing of DB code. + + Using `DbFixture`. Intended to be the main database test case to use all + the tests on a given backend with user defined uri. Backend specific + tests should be decorated with `backend_specific` decorator. + """ + + FIXTURE = DbFixture + + def setUp(self): + super(DbTestCase, self).setUp() + self.useFixture(self.FIXTURE(self)) + + +ALLOWED_DIALECTS = ['sqlite', 'mysql', 'postgresql'] + + +def backend_specific(*dialects): + """Decorator to skip backend specific tests on inappropriate engines. + + ::dialects: list of dialects names under which the test will be launched. + """ + def wrap(f): + @functools.wraps(f) + def ins_wrap(self): + if not set(dialects).issubset(ALLOWED_DIALECTS): + raise ValueError( + "Please use allowed dialects: %s" % ALLOWED_DIALECTS) + if self.engine.name not in dialects: + msg = ('The test "%s" can be run ' + 'only on %s. Current engine is %s.') + args = (f.__name__, ' '.join(dialects), self.engine.name) + self.skip(msg % args) + else: + return f(self) + return ins_wrap + return wrap + + +@six.add_metaclass(abc.ABCMeta) +class OpportunisticFixture(DbFixture): + """Base fixture to use default CI databases. + + The databases exist in OpenStack CI infrastructure. But for the + correct functioning in local environment the databases must be + created manually. + """ + + DRIVER = abc.abstractproperty(lambda: None) + DBNAME = PASSWORD = USERNAME = 'openstack_citest' + + def setUp(self): + self._provisioning_engine = provision.get_engine( + utils.get_connect_string(backend=self.DRIVER, + user=self.USERNAME, + passwd=self.PASSWORD, + database=self.DBNAME) + ) + self._uri = provision.create_database(self._provisioning_engine) + + super(OpportunisticFixture, self).setUp() + + def cleanUp(self): + super(OpportunisticFixture, self).cleanUp() + + provision.drop_database(self._provisioning_engine, self._uri) + + def _get_uri(self): + return self._uri + + +@six.add_metaclass(abc.ABCMeta) +class OpportunisticTestCase(DbTestCase): + """Base test case to use default CI databases. + + The subclasses of the test case are running only when openstack_citest + database is available otherwise a tests will be skipped. + """ + + FIXTURE = abc.abstractproperty(lambda: None) + + def setUp(self): + credentials = { + 'backend': self.FIXTURE.DRIVER, + 'user': self.FIXTURE.USERNAME, + 'passwd': self.FIXTURE.PASSWORD, + 'database': self.FIXTURE.DBNAME} + + if self.FIXTURE.DRIVER and not utils.is_backend_avail(**credentials): + msg = '%s backend is not available.' % self.FIXTURE.DRIVER + return self.skip(msg) + + super(OpportunisticTestCase, self).setUp() + + +class MySQLOpportunisticFixture(OpportunisticFixture): + DRIVER = 'mysql' + + +class PostgreSQLOpportunisticFixture(OpportunisticFixture): + DRIVER = 'postgresql' + + +class MySQLOpportunisticTestCase(OpportunisticTestCase): + FIXTURE = MySQLOpportunisticFixture + + +class PostgreSQLOpportunisticTestCase(OpportunisticTestCase): + FIXTURE = PostgreSQLOpportunisticFixture diff --git a/sahara/openstack/common/db/sqlalchemy/test_migrations.py b/sahara/openstack/common/db/sqlalchemy/test_migrations.py index 9626a0cc32..194c381850 100644 --- a/sahara/openstack/common/db/sqlalchemy/test_migrations.py +++ b/sahara/openstack/common/db/sqlalchemy/test_migrations.py @@ -15,19 +15,19 @@ # under the License. import functools +import logging import os import subprocess import lockfile +from oslotest import base as test_base from six import moves +from six.moves.urllib import parse import sqlalchemy import sqlalchemy.exc from sahara.openstack.common.db.sqlalchemy import utils -from sahara.openstack.common.gettextutils import _ -from sahara.openstack.common import log as logging -from sahara.openstack.common.py3kcompat import urlutils -from sahara.openstack.common import test +from sahara.openstack.common.gettextutils import _LE LOG = logging.getLogger(__name__) @@ -60,15 +60,15 @@ def _set_db_lock(lock_path=None, lock_prefix=None): path = lock_path or os.environ.get("SAHARA_LOCK_PATH") lock = lockfile.FileLock(os.path.join(path, lock_prefix)) with lock: - LOG.debug(_('Got lock "%s"') % f.__name__) + LOG.debug('Got lock "%s"' % f.__name__) return f(*args, **kwargs) finally: - LOG.debug(_('Lock released "%s"') % f.__name__) + LOG.debug('Lock released "%s"' % f.__name__) return wrapper return decorator -class BaseMigrationTestCase(test.BaseTestCase): +class BaseMigrationTestCase(test_base.BaseTestCase): """Base class fort testing of migration utils.""" def __init__(self, *args, **kwargs): @@ -153,7 +153,7 @@ class BaseMigrationTestCase(test.BaseTestCase): def _reset_databases(self): for key, engine in self.engines.items(): conn_string = self.test_databases[key] - conn_pieces = urlutils.urlparse(conn_string) + conn_pieces = parse.urlparse(conn_string) engine.dispose() if conn_string.startswith('sqlite'): # We can just delete the SQLite database, which is @@ -264,6 +264,6 @@ class WalkVersionsMixin(object): if check: check(engine, data) except Exception: - LOG.error("Failed to migrate to version %s on engine %s" % + LOG.error(_LE("Failed to migrate to version %s on engine %s") % (version, engine)) raise diff --git a/sahara/openstack/common/db/sqlalchemy/utils.py b/sahara/openstack/common/db/sqlalchemy/utils.py index 5827b685d1..7fae28f43b 100644 --- a/sahara/openstack/common/db/sqlalchemy/utils.py +++ b/sahara/openstack/common/db/sqlalchemy/utils.py @@ -16,9 +16,9 @@ # License for the specific language governing permissions and limitations # under the License. +import logging import re -from migrate.changeset import UniqueConstraint import sqlalchemy from sqlalchemy import Boolean from sqlalchemy import CheckConstraint @@ -29,16 +29,16 @@ from sqlalchemy import func from sqlalchemy import Index from sqlalchemy import Integer from sqlalchemy import MetaData +from sqlalchemy import or_ from sqlalchemy.sql.expression import literal_column from sqlalchemy.sql.expression import UpdateBase -from sqlalchemy.sql import select from sqlalchemy import String from sqlalchemy import Table from sqlalchemy.types import NullType -from sahara.openstack.common.gettextutils import _ - -from sahara.openstack.common import log as logging +from sahara.openstack.common import context as request_context +from sahara.openstack.common.db.sqlalchemy import models +from sahara.openstack.common.gettextutils import _, _LI, _LW from sahara.openstack.common import timeutils @@ -94,7 +94,7 @@ def paginate_query(query, model, limit, sort_keys, marker=None, if 'id' not in sort_keys: # TODO(justinsb): If this ever gives a false-positive, check # the actual primary key, rather than assuming its id - LOG.warning(_('Id not in sort_keys; is sort_keys unique?')) + LOG.warning(_LW('Id not in sort_keys; is sort_keys unique?')) assert(not (sort_dir and sort_dirs)) @@ -157,11 +157,111 @@ def paginate_query(query, model, limit, sort_keys, marker=None, return query +def _read_deleted_filter(query, db_model, read_deleted): + if 'deleted' not in db_model.__table__.columns: + raise ValueError(_("There is no `deleted` column in `%s` table. " + "Project doesn't use soft-deleted feature.") + % db_model.__name__) + + default_deleted_value = db_model.__table__.c.deleted.default.arg + if read_deleted == 'no': + query = query.filter(db_model.deleted == default_deleted_value) + elif read_deleted == 'yes': + pass # omit the filter to include deleted and active + elif read_deleted == 'only': + query = query.filter(db_model.deleted != default_deleted_value) + else: + raise ValueError(_("Unrecognized read_deleted value '%s'") + % read_deleted) + return query + + +def _project_filter(query, db_model, context, project_only): + if project_only and 'project_id' not in db_model.__table__.columns: + raise ValueError(_("There is no `project_id` column in `%s` table.") + % db_model.__name__) + + if request_context.is_user_context(context) and project_only: + if project_only == 'allow_none': + is_none = None + query = query.filter(or_(db_model.project_id == context.project_id, + db_model.project_id == is_none)) + else: + query = query.filter(db_model.project_id == context.project_id) + + return query + + +def model_query(context, model, session, args=None, project_only=False, + read_deleted=None): + """Query helper that accounts for context's `read_deleted` field. + + :param context: context to query under + + :param model: Model to query. Must be a subclass of ModelBase. + :type model: models.ModelBase + + :param session: The session to use. + :type session: sqlalchemy.orm.session.Session + + :param args: Arguments to query. If None - model is used. + :type args: tuple + + :param project_only: If present and context is user-type, then restrict + query to match the context's project_id. If set to + 'allow_none', restriction includes project_id = None. + :type project_only: bool + + :param read_deleted: If present, overrides context's read_deleted field. + :type read_deleted: bool + + Usage: + + ..code:: python + + result = (utils.model_query(context, models.Instance, session=session) + .filter_by(uuid=instance_uuid) + .all()) + + query = utils.model_query( + context, Node, + session=session, + args=(func.count(Node.id), func.sum(Node.ram)) + ).filter_by(project_id=project_id) + + """ + + if not read_deleted: + if hasattr(context, 'read_deleted'): + # NOTE(viktors): some projects use `read_deleted` attribute in + # their contexts instead of `show_deleted`. + read_deleted = context.read_deleted + else: + read_deleted = context.show_deleted + + if not issubclass(model, models.ModelBase): + raise TypeError(_("model should be a subclass of ModelBase")) + + query = session.query(model) if not args else session.query(*args) + query = _read_deleted_filter(query, model, read_deleted) + query = _project_filter(query, model, context, project_only) + + return query + + def get_table(engine, name): """Returns an sqlalchemy table dynamically from db. Needed because the models don't work for us in migrations as models will be far out of sync with the current data. + + .. warning:: + + Do not use this method when creating ForeignKeys in database migrations + because sqlalchemy needs the same MetaData object to hold information + about the parent table and the reference table in the ForeignKey. This + method uses a unique MetaData object per table object so it won't work + with ForeignKey creation. """ metadata = MetaData() metadata.bind = engine @@ -208,6 +308,10 @@ def drop_unique_constraint(migrate_engine, table_name, uc_name, *columns, **col_name_col_instance): """Drop unique constraint from table. + DEPRECATED: this function is deprecated and will be removed from sahara.db + in a few releases. Please use UniqueConstraint.drop() method directly for + sqlalchemy-migrate migration scripts. + This method drops UC from table and works for mysql, postgresql and sqlite. In mysql and postgresql we are able to use "alter table" construction. Sqlalchemy doesn't support some sqlite column types and replaces their @@ -224,6 +328,8 @@ def drop_unique_constraint(migrate_engine, table_name, uc_name, *columns, types by sqlite. For example BigInteger. """ + from migrate.changeset import UniqueConstraint + meta = MetaData() meta.bind = migrate_engine t = Table(table_name, meta, autoload=True) @@ -263,9 +369,9 @@ def drop_old_duplicate_entries_from_table(migrate_engine, table_name, columns_for_select = [func.max(table.c.id)] columns_for_select.extend(columns_for_group_by) - duplicated_rows_select = select(columns_for_select, - group_by=columns_for_group_by, - having=func.count(table.c.id) > 1) + duplicated_rows_select = sqlalchemy.sql.select( + columns_for_select, group_by=columns_for_group_by, + having=func.count(table.c.id) > 1) for row in migrate_engine.execute(duplicated_rows_select): # NOTE(boris-42): Do not remove row that has the biggest ID. @@ -275,10 +381,11 @@ def drop_old_duplicate_entries_from_table(migrate_engine, table_name, for name in uc_column_names: delete_condition &= table.c[name] == row[name] - rows_to_delete_select = select([table.c.id]).where(delete_condition) + rows_to_delete_select = sqlalchemy.sql.select( + [table.c.id]).where(delete_condition) for row in migrate_engine.execute(rows_to_delete_select).fetchall(): - LOG.info(_("Deleting duplicated row with id: %(id)s from table: " - "%(table)s") % dict(id=row[0], table=table_name)) + LOG.info(_LI("Deleting duplicated row with id: %(id)s from table: " + "%(table)s") % dict(id=row[0], table=table_name)) if use_soft_delete: delete_statement = table.update().\ @@ -386,7 +493,7 @@ def _change_deleted_column_type_to_boolean_sqlite(migrate_engine, table_name, else: c_select.append(table.c.deleted == table.c.id) - ins = InsertFromSelect(new_table, select(c_select)) + ins = InsertFromSelect(new_table, sqlalchemy.sql.select(c_select)) migrate_engine.execute(ins) table.drop() diff --git a/sahara/tests/unit/base.py b/sahara/tests/unit/base.py index 69a73dec31..8b8cad13fe 100644 --- a/sahara/tests/unit/base.py +++ b/sahara/tests/unit/base.py @@ -22,7 +22,7 @@ import unittest2 from sahara import context from sahara.db import api as db_api from sahara import main -from sahara.openstack.common.db.sqlalchemy import session +from sahara.openstack.common.db import options class SaharaTestCase(unittest2.TestCase): @@ -55,7 +55,7 @@ class SaharaWithDbTestCase(SaharaTestCase): def setup_db(self): self.db_fd, self.db_path = tempfile.mkstemp() - session.set_defaults('sqlite:///' + self.db_path, self.db_path) + options.set_defaults('sqlite:///' + self.db_path, self.db_path) db_api.setup_db() self.addCleanup(self._drop_db) diff --git a/sahara/tests/unit/db/migration/test_migrations_base.py b/sahara/tests/unit/db/migration/test_migrations_base.py index 0568a9d0f6..a581055f88 100644 --- a/sahara/tests/unit/db/migration/test_migrations_base.py +++ b/sahara/tests/unit/db/migration/test_migrations_base.py @@ -36,7 +36,7 @@ import sqlalchemy.exc import unittest2 import sahara.db.migration -from sahara.openstack.common.db.sqlalchemy import session +from sahara.db.sqlalchemy import api as sa from sahara.openstack.common import lockutils from sahara.openstack.common import log as logging from sahara.openstack.common import processutils @@ -394,7 +394,7 @@ class BaseWalkMigrationTestCase(BaseMigrationTestCase): database functionality (reset default settings and session cleanup). """ CONF.set_override('connection', str(engine.url), group='database') - session.cleanup() + sa.cleanup() def _test_mysql_opportunistically(self): # Test that table creation on mysql only builds InnoDB tables @@ -453,11 +453,11 @@ class BaseWalkMigrationTestCase(BaseMigrationTestCase): """ self.ALEMBIC_CONFIG.stdout = buf = io.StringIO() CONF.set_override('connection', str(engine.url), group='database') - session.cleanup() + sa.cleanup() getattr(command, alembic_command)(*args, **kwargs) res = buf.getvalue().strip() LOG.debug('Alembic command `%s` returns: %s' % (alembic_command, res)) - session.cleanup() + sa.cleanup() return res def _get_alembic_versions(self, engine): diff --git a/test-requirements.txt b/test-requirements.txt index 2d884712bf..b13890eae1 100644 --- a/test-requirements.txt +++ b/test-requirements.txt @@ -7,6 +7,7 @@ docutils==0.9.1 fixtures>=0.3.14 mock>=1.0 oslosphinx +oslotest psycopg2 pylint==0.25.2 sphinx>=1.2.1,<1.3