Merge "Move DB thread pooling to DB API loader"

4 files changed, 188 insertions, 75 deletions

diff --git a/openstack/common/db/api.py b/openstack/common/db/api.py
new file mode 100644
index 0000000..407790d
--- /dev/null
+++ b/openstack/common/db/api.py
@@ -0,0 +1,101 @@
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright (c) 2013 Rackspace Hosting
+# 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.
+
+"""Multiple DB API backend support.
+
+Supported configuration options:
+
+`db_backend`: DB backend name or full module path to DB backend module.
+`dbapi_use_tpool`: Enable thread pooling of DB API calls.
+
+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.
+
+*NOTE*: There are bugs in eventlet when using tpool combined with
+threading locks. The python logging module happens to use such locks.  To
+work around this issue, be sure to specify thread=False with
+eventlet.monkey_patch().
+
+A bug for eventlet has been filed here:
+
+https://bitbucket.org/eventlet/eventlet/issue/137/
+"""
+import functools
+
+from oslo.config import cfg
+
+from openstack.common import lockutils
+from openstack.common import importutils
+
+
+db_opts = [
+    cfg.StrOpt('db_backend',
+               default='sqlalchemy',
+               help='The backend to use for db'),
+    cfg.BoolOpt('dbapi_use_tpool',
+                default=False,
+                help='Enable the experimental use of thread pooling for '
+                     'all DB API calls')
+]
+
+CONF = cfg.CONF
+CONF.register_opts(db_opts)
+
+
+class DBAPI(object):
+    def __init__(self, backend_mapping=None):
+        if backend_mapping is None:
+            backend_mapping = {}
+        self.__backend = None
+        self.__backend_mapping = backend_mapping
+
+    @lockutils.synchronized('dbapi_backend', 'oslo-')
+    def __get_backend(self):
+        """Get the actual backend.  May be a module or an instance of
+        a class.  Doesn't matter to us.  We do this synchronized as it's
+        possible multiple greenthreads started very quickly trying to do
+        DB calls and eventlet can switch threads before self.__backend gets
+        assigned.
+        """
+        if self.__backend:
+            # Another thread assigned it
+            return self.__backend
+        backend_name = CONF.db_backend
+        self.__use_tpool = CONF.dbapi_use_tpool
+        if self.__use_tpool:
+            from eventlet import tpool
+            self.__tpool = tpool
+        # Import the untranslated name if we don't have a
+        # mapping.
+        backend_path = self.__backend_mapping.get(backend_name,
+                                                  backend_name)
+        backend_mod = importutils.import_module(backend_path)
+        self.__backend = backend_mod.get_backend()
+        return self.__backend
+
+    def __getattr__(self, key):
+        backend = self.__backend or self.__get_backend()
+        attr = getattr(backend, key)
+        if not self.__use_tpool or not hasattr(attr, '__call__'):
+            return attr
+
+        def tpool_wrapper(*args, **kwargs):
+            return self.__tpool.execute(attr, *args, **kwargs)
+
+        functools.update_wrapper(tpool_wrapper, attr)
+        return tpool_wrapper
diff --git a/openstack/common/db/sqlalchemy/session.py b/openstack/common/db/sqlalchemy/session.py
index 1e9e123..96f582f 100644
--- a/openstack/common/db/sqlalchemy/session.py
+++ b/openstack/common/db/sqlalchemy/session.py
@@ -244,7 +244,6 @@ import os.path
 import re
 import time
 
-from eventlet import db_pool
 from eventlet import greenthread
 from oslo.config import cfg
 from sqlalchemy.exc import DisconnectionError, OperationalError, IntegrityError
@@ -253,14 +252,10 @@ import sqlalchemy.orm
 from sqlalchemy.pool import NullPool, StaticPool
 from sqlalchemy.sql.expression import literal_column
 
-from openstack.common import importutils
 from openstack.common import log as logging
 from openstack.common.gettextutils import _
 from openstack.common import timeutils
 
-MySQLdb = importutils.try_import('MySQLdb')
-if MySQLdb is not None:
-    from MySQLdb.constants import CLIENT as mysql_client_constants
 
 sql_opts = [
     cfg.StrOpt('sql_connection',
@@ -303,9 +298,6 @@ sql_opts = [
     cfg.BoolOpt('sql_connection_trace',
                 default=False,
                 help='Add python stack traces to SQL as comment strings'),
-    cfg.BoolOpt('sql_dbpool_enable',
-                default=False,
-                help="enable the use of eventlet's db_pool for MySQL"),
 ]
 
 CONF = cfg.CONF
@@ -517,33 +509,6 @@ def create_engine(sql_connection):
         if CONF.sql_connection == "sqlite://":
             engine_args["poolclass"] = StaticPool
             engine_args["connect_args"] = {'check_same_thread': False}
-    elif all((CONF.sql_dbpool_enable, MySQLdb,
-              "mysql" in connection_dict.drivername)):
-        LOG.info(_("Using mysql/eventlet db_pool."))
-        # MySQLdb won't accept 'None' in the password field
-        password = connection_dict.password or ''
-        pool_args = {
-            'db': connection_dict.database,
-            'passwd': password,
-            'host': connection_dict.host,
-            'user': connection_dict.username,
-            'min_size': CONF.sql_min_pool_size,
-            'max_size': CONF.sql_max_pool_size,
-            'max_idle': CONF.sql_idle_timeout,
-            'client_flag': mysql_client_constants.FOUND_ROWS}
-
-        pool = db_pool.ConnectionPool(MySQLdb, **pool_args)
-
-        def creator():
-            conn = pool.create()
-            if isinstance(conn, tuple):
-                # NOTE(belliott) eventlet >= 0.10 returns a tuple
-                now, now, conn = conn
-
-            return conn
-
-        engine_args['creator'] = creator
-
     else:
         engine_args['pool_size'] = CONF.sql_max_pool_size
         if CONF.sql_max_overflow is not None:
diff --git a/tests/unit/db/sqlalchemy/test_sqlalchemy.py b/tests/unit/db/sqlalchemy/test_sqlalchemy.py
index c063f0d..6b84f10 100644
--- a/tests/unit/db/sqlalchemy/test_sqlalchemy.py
+++ b/tests/unit/db/sqlalchemy/test_sqlalchemy.py
@@ -16,8 +16,6 @@
 
 """Unit tests for SQLAlchemy specific code."""
 
-from eventlet import db_pool
-
 from sqlalchemy import Column, MetaData, Table, UniqueConstraint
 from sqlalchemy.ext.declarative import declarative_base
 from sqlalchemy import DateTime, Integer
@@ -34,44 +32,6 @@ class TestException(Exception):
     pass
 
 
-class DbPoolTestCase(test_utils.BaseTestCase):
-    def setUp(self):
-        super(DbPoolTestCase, self).setUp()
-        if MySQLdb is None:
-            self.skipTest("Required module MySQLdb missing.")
-        self.config(sql_dbpool_enable=True)
-        self.user_id = 'fake'
-        self.project_id = 'fake'
-
-    def test_db_pool_option(self):
-        self.config(sql_idle_timeout=11, sql_min_pool_size=21,
-                    sql_max_pool_size=42)
-
-        info = {}
-
-        class FakeConnectionPool(db_pool.ConnectionPool):
-            def __init__(self, mod_name, **kwargs):
-                info['module'] = mod_name
-                info['kwargs'] = kwargs
-                super(FakeConnectionPool, self).__init__(mod_name,
-                                                         **kwargs)
-
-            def connect(self, *args, **kwargs):
-                raise TestException()
-
-        self.stubs.Set(db_pool, 'ConnectionPool',
-                       FakeConnectionPool)
-
-        sql_connection = 'mysql://user:pass@127.0.0.1/nova'
-        self.assertRaises(TestException, session.create_engine,
-                          sql_connection)
-
-        self.assertEqual(info['module'], MySQLdb)
-        self.assertEqual(info['kwargs']['max_idle'], 11)
-        self.assertEqual(info['kwargs']['min_size'], 21)
-        self.assertEqual(info['kwargs']['max_size'], 42)
-
-
 BASE = declarative_base()
 _TABLE_NAME = '__tmp__test__tmp__'
 
diff --git a/tests/unit/db/test_api.py b/tests/unit/db/test_api.py
new file mode 100644
index 0000000..a31ffd0
--- /dev/null
+++ b/tests/unit/db/test_api.py
@@ -0,0 +1,87 @@
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+# Copyright (c) 2013 Rackspace Hosting
+# 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.
+
+"""Unit tests for DB API."""
+
+from eventlet import tpool
+
+from openstack.common.db import api
+from tests import utils as test_utils
+
+
+def get_backend():
+    return DBAPI()
+
+
+class DBAPI(object):
+    def api_class_call1(_self, *args, **kwargs):
+        return args, kwargs
+
+
+class DBAPITestCase(test_utils.BaseTestCase):
+    def test_dbapi_api_class_method_and_tpool_false(self):
+        backend_mapping = {'test_known': 'tests.unit.db.test_api'}
+        self.config(db_backend='test_known', dbapi_use_tpool=False)
+
+        info = dict(tpool=False)
+        orig_execute = tpool.execute
+
+        def our_execute(*args, **kwargs):
+            info['tpool'] = True
+            return orig_execute(*args, **kwargs)
+
+        self.stubs.Set(tpool, 'execute', our_execute)
+
+        dbapi = api.DBAPI(backend_mapping=backend_mapping)
+        result = dbapi.api_class_call1(1, 2, kwarg1='meow')
+        expected = ((1, 2), {'kwarg1': 'meow'})
+        self.assertEqual(expected, result)
+        self.assertFalse(info['tpool'])
+
+    def test_dbapi_api_class_method_and_tpool_true(self):
+        backend_mapping = {'test_known': 'tests.unit.db.test_api'}
+        self.config(db_backend='test_known', dbapi_use_tpool=True)
+
+        info = dict(tpool=False)
+        orig_execute = tpool.execute
+
+        def our_execute(*args, **kwargs):
+            info['tpool'] = True
+            return orig_execute(*args, **kwargs)
+
+        self.stubs.Set(tpool, 'execute', our_execute)
+
+        dbapi = api.DBAPI(backend_mapping=backend_mapping)
+        result = dbapi.api_class_call1(1, 2, kwarg1='meow')
+        expected = ((1, 2), {'kwarg1': 'meow'})
+        self.assertEqual(expected, result)
+        self.assertTrue(info['tpool'])
+
+    def test_dbapi_full_path_module_method(self):
+        self.config(db_backend='tests.unit.db.test_api')
+        dbapi = api.DBAPI()
+        result = dbapi.api_class_call1(1, 2, kwarg1='meow')
+        expected = ((1, 2), {'kwarg1': 'meow'})
+        self.assertEqual(expected, result)
+
+    def test_dbapi_unknown_invalid_backend(self):
+        self.config(db_backend='tests.unit.db.not_existant')
+        dbapi = api.DBAPI()
+
+        def call_it():
+            dbapi.api_class_call1(1, 2, kwarg1='meow')
+
+        self.assertRaises(ImportError, call_it)