Fixed sqlalchemy 2.x support

Re-arranged sessions to always be at the top level to make
it easier to know the origin of the current session. This
is important now that we no longer have autocommit enabled.

- Added zuul job testing sqlalchemy 2.x.
- Added new db api for service cleanup.
- Removed broken sqlite cleanup step during testing.

Change-Id: I168f3d9518611ac66cb9eec1132a7add19e92d5f

.zuul.yaml

@@ -8,8 +8,10 @@
     check:
       jobs:
         - senlin-dsvm-tempest-py3-api
+        - senlin-dsvm-tempest-py3-api-sqlalchemy-2x
         - senlin-tempest-api-ipv6-only
         - senlin-dsvm-tempest-py3-functional
+        - senlin-dsvm-tempest-py3-functional-sqlalchemy-2x
         - senlin-dsvm-tempest-py3-integration
         - senlin-dsvm-tempest-py3-integration-zaqar:
             voting: false
@@ -65,6 +67,22 @@
              DEFAULT:
                cloud_backend: openstack_test
 
+- job:
+    name: senlin-dsvm-tempest-py3-api-sqlalchemy-2x
+    parent: senlin-tempest-base
+    required-projects:
+      - name: openstack/oslo.db
+    vars:
+      tempest_test_regex: senlin_tempest_plugin.tests.api
+      devstack_localrc:
+        USE_PYTHON3: true
+        USE_SQLALCHEMY_LATEST: true
+      devstack_local_conf:
+        post-config:
+          $SENLIN_CONF:
+             DEFAULT:
+               cloud_backend: openstack_test
+
 - job:
     name: senlin-dsvm-tempest-py3-functional
     parent: senlin-tempest-base
@@ -79,6 +97,23 @@
                cloud_backend: openstack_test
                health_check_interval_min: 10
 
+- job:
+    name: senlin-dsvm-tempest-py3-functional-sqlalchemy-2x
+    parent: senlin-tempest-base
+    required-projects:
+      - name: openstack/oslo.db
+    vars:
+      tempest_test_regex: senlin_tempest_plugin.tests.functional
+      devstack_localrc:
+        USE_PYTHON3: true
+        USE_SQLALCHEMY_LATEST: true
+      devstack_local_conf:
+        post-config:
+          $SENLIN_CONF:
+             DEFAULT:
+               cloud_backend: openstack_test
+               health_check_interval_min: 10
+
 - job:
     name: senlin-dsvm-tempest-py3-integration
     parent: senlin-tempest-base
@@ -143,4 +178,3 @@
           $SENLIN_CONF:
              DEFAULT:
                cloud_backend: openstack_test
-

devstack/lib/senlin

@@ -196,10 +196,13 @@ function _config_senlin_apache_wsgi {
 
 # init_senlin() - Initialize database
 function init_senlin {
-
     # (re)create senlin database
     recreate_database senlin utf8
 
+    if [[ "$USE_SQLALCHEMY_LATEST" == "True" ]]; then
+        pip3 install --upgrade alembic sqlalchemy
+    fi
+
     $SENLIN_BIN_DIR/senlin-manage db_sync
     create_senlin_cache_dir
 }

releasenotes/notes/Updated-for-SQLAlchemy-2.x-ee6831e5a95d3658.yaml

@@ -0,0 +1,4 @@
+---
+fixes:
+  - |
+    Fixed compatibility issues with SQLAlchemy 2.x.

senlin/cmd/status.py

@@ -20,6 +20,7 @@ from oslo_upgradecheck import upgradecheck
 
 from senlin.common.i18n import _
 from senlin.db import api
+from senlin.db.sqlalchemy import api as sql_api
 
 from sqlalchemy import MetaData, Table, select, column
 
@@ -42,15 +43,19 @@ class Checks(upgradecheck.UpgradeCommands):
         """
 
         engine = api.get_engine()
-        metadata = MetaData(bind=engine)
+        metadata = MetaData()
-        policy = Table('policy', metadata, autoload=True)
+        metadata.bind = engine
+
+        policy = Table('policy', metadata, autoload_with=engine)
 
         healthpolicy_select = (
-            select([column('name')])
+            select(column('name'))
             .select_from(policy)
             .where(column('type') == 'senlin.policy.health-1.0')
         )
-        healthpolicy_rows = engine.execute(healthpolicy_select).fetchall()
+
+        with sql_api.session_for_read() as session:
+            healthpolicy_rows = session.execute(healthpolicy_select).fetchall()
 
         if not healthpolicy_rows:
             return upgradecheck.Result(upgradecheck.Code.SUCCESS)

senlin/common/service.py

@@ -90,29 +90,7 @@ class Service(service.Service):
     def service_manage_cleanup(self):
         self.cleanup_count += 1
         try:
-            ctx = senlin_context.get_admin_context()
+            service_obj.Service.cleanup_all_expired(self.name)
-            services = service_obj.Service.get_all_expired(
-                ctx, self.name
-            )
-            for svc in services:
-                LOG.info(
-                    'Breaking locks for dead service %(name)s '
-                    '(id: %(service_id)s)',
-                    {
-                        'name': self.name,
-                        'service_id': svc['id'],
-                    }
-                )
-                service_obj.Service.gc_by_engine(svc['id'])
-                LOG.info(
-                    'Done breaking locks for service %(name)s '
-                    '(id: %(service_id)s)',
-                    {
-                        'name': self.name,
-                        'service_id': svc['id'],
-                    }
-                )
-                service_obj.Service.delete(svc['id'])
         except Exception as ex:
             LOG.error(
                 'Error while cleaning up service %(name)s: %(ex)s',

senlin/db/api.py

@@ -486,8 +486,8 @@ def service_get_all():
     return IMPL.service_get_all()
 
 
-def service_get_all_expired(binary):
+def service_cleanup_all_expired(binary):
-    return IMPL.service_get_all_expired(binary)
+    return IMPL.service_cleanup_all_expired(binary)
 
 
 def gc_by_engine(engine_id):

senlin/objects/service.py

@@ -49,9 +49,8 @@ class Service(base.SenlinObject, base.VersionedObjectDictCompat):
         return [cls._from_db_object(context, cls(), obj) for obj in objs]
 
     @classmethod
-    def get_all_expired(cls, context, binary):
+    def cleanup_all_expired(cls, binary):
-        objs = db_api.service_get_all_expired(binary)
+        db_api.service_cleanup_all_expired(binary)
-        return [cls._from_db_object(context, cls(), obj) for obj in objs]
 
     @classmethod
     def update(cls, context, obj_id, values=None):

senlin/tests/unit/common/base.py

@@ -78,7 +78,9 @@ class DatabaseFixture(fixtures.Fixture):
         super(DatabaseFixture, self).__init__()
         self.golden_path = self.mktemp()
         self.golden_url = 'sqlite:///%s' % self.golden_path
+
         db_api.db_sync(self.golden_url)
+
         self.working_path = self.mktemp()
         self.working_url = 'sqlite:///%s' % self.working_path
 
@@ -86,10 +88,6 @@ class DatabaseFixture(fixtures.Fixture):
         super(DatabaseFixture, self).setUp()
         shutil.copy(self.golden_path, self.working_path)
 
-    def cleanup(self):
-        if os.path.exists(self.working_path):
-            os.remove(self.working_path)
-
 
 class SenlinTestCase(testscenarios.WithScenarios,
                      testtools.TestCase, FakeLogMixin):
@@ -114,7 +112,6 @@ class SenlinTestCase(testscenarios.WithScenarios,
         self.addCleanup(messaging.cleanup)
 
         self.db_fixture = self.useFixture(DatabaseFixture.get_fixture())
-        self.addCleanup(self.db_fixture.cleanup)
 
         options.cfg.set_defaults(
             options.database_opts, sqlite_synchronous=False

senlin/tests/unit/conductor/test_service.py

@@ -10,13 +10,11 @@
 # License for the specific language governing permissions and limitations
 # under the License.
 
-import datetime
 from unittest import mock
 
 import eventlet
 from oslo_config import cfg
 import oslo_messaging
-from oslo_utils import timeutils
 from oslo_utils import uuidutils
 
 from senlin.common import consts
@@ -151,25 +149,16 @@ class ConductorCleanupTest(base.SenlinTestCase):
         self.assertGreater(mock_update.call_count, 1)
         self.svc.stop()
 
-    @mock.patch.object(service_obj.Service, 'gc_by_engine')
+    @mock.patch.object(service_obj.Service, 'cleanup_all_expired')
-    @mock.patch.object(service_obj.Service, 'get_all_expired')
+    def test_service_manage_cleanup(self, mock_cleanup):
-    @mock.patch.object(service_obj.Service, 'delete')
-    def test_service_manage_cleanup(self, mock_delete, mock_get_all_expired,
-                                    mock_gc):
         self.svc = service.ConductorService('HOST', self.topic)
         self.svc.service_id = self.service_id
-        delta = datetime.timedelta(seconds=2.2 * cfg.CONF.periodic_interval)
-        ages_a_go = timeutils.utcnow(True) - delta
-        mock_get_all_expired.return_value = [
-            {'id': 'foo', 'updated_at': ages_a_go}
-        ]
         self.svc.service_manage_cleanup()
-        mock_delete.assert_called_once_with('foo')
+        mock_cleanup.assert_called_once_with('senlin-conductor')
-        mock_gc.assert_called_once_with('foo')
 
-    @mock.patch.object(service_obj.Service, 'get_all_expired')
+    @mock.patch.object(service_obj.Service, 'cleanup_all_expired')
     def test_service_manage_cleanup_without_exception(self,
-                                                      mock_get_all_expired):
+                                                      mock_cleanup):
         cfg.CONF.set_override('periodic_interval', 0.1)
 
         self.svc = service.ConductorService('HOST', self.topic)
@@ -178,11 +167,11 @@ class ConductorCleanupTest(base.SenlinTestCase):
         # start engine and verify that get_all is being called more than once
         self.svc.start()
         eventlet.sleep(0.6)
-        self.assertGreater(mock_get_all_expired.call_count, 1)
         self.svc.stop()
+        mock_cleanup.assert_called()
 
-    @mock.patch.object(service_obj.Service, 'get_all_expired')
+    @mock.patch.object(service_obj.Service, 'cleanup_all_expired')
-    def test_service_manage_cleanup_with_exception(self, mock_get_all_expired):
+    def test_service_manage_cleanup_with_exception(self, mock_cleanup):
         cfg.CONF.set_override('periodic_interval', 0.1)
 
         self.svc = service.ConductorService('HOST', self.topic)
@@ -190,8 +179,8 @@ class ConductorCleanupTest(base.SenlinTestCase):
 
         # start engine and verify that get_all is being called more than once
         # even with the exception being thrown
-        mock_get_all_expired.side_effect = Exception('blah')
+        mock_cleanup.side_effect = Exception('blah')
         self.svc.start()
         eventlet.sleep(0.6)
-        self.assertGreater(mock_get_all_expired.call_count, 1)
         self.svc.stop()
+        mock_cleanup.assert_called()

senlin/tests/unit/db/test_service_api.py

@@ -15,7 +15,6 @@ from oslo_utils import timeutils
 from oslo_utils import uuidutils
 
 from senlin.db.sqlalchemy import api as db_api
-from senlin.db.sqlalchemy import models
 from senlin.tests.unit.common import base
 from senlin.tests.unit.common import utils
 
@@ -34,19 +33,12 @@ class DBAPIServiceTest(base.SenlinTestCase):
         }
         values.update(kwargs)
 
-        with db_api.session_for_write() as session:
+        return db_api.service_create(
-            time_now = timeutils.utcnow(True)
+            service_id, host=values.get('host'),
-            svc = models.Service(
+            binary=values.get('binary'),
-                id=service_id,
+            topic=values.get('topic'),
-                host=values.get('host'),
+            time_now=kwargs.get('time_now')
-                binary=values.get('binary'),
+        )
-                topic=values.get('topic'),
-                created_at=values.get('created_at') or time_now,
-                updated_at=values.get('updated_at') or time_now,
-            )
-            session.add(svc)
-
-        return svc
 
     def test_service_create_get(self):
         service = self._create_service()
@@ -74,32 +66,31 @@ class DBAPIServiceTest(base.SenlinTestCase):
         self.assertEqual(4, len(services))
 
     def test_service_get_all_expired(self):
-        for index in range(6):
+        for index in range(3):
-            dt = timeutils.utcnow() - datetime.timedelta(seconds=60 * index)
+            dt = timeutils.utcnow() - datetime.timedelta(hours=8)
             values = {
                 'binary': 'senlin-health-manager',
-                'host': 'host-%s' % index,
+                'host': 'host-0-%s' % index,
-                'updated_at': dt
+                'time_now': dt
             }
             self._create_service(uuidutils.generate_uuid(), **values)
 
-        for index in range(8):
+        for index in range(3):
-            dt = timeutils.utcnow() - datetime.timedelta(seconds=60 * index)
+            dt = timeutils.utcnow()
             values = {
-                'binary': 'senlin-engine',
+                'binary': 'senlin-health-manager',
-                'host': 'host-%s' % index,
+                'host': 'host-1-%s' % index,
-                'updated_at': dt
+                'time_now': dt
             }
             self._create_service(uuidutils.generate_uuid(), **values)
 
-        services = db_api.service_get_all_expired('senlin-health-manager')
+        db_api.service_cleanup_all_expired('senlin-health-manager')
-        self.assertEqual(3, len(services.all()))
 
-        services = db_api.service_get_all_expired('senlin-engine')
+        self.assertEqual(3, len(db_api.service_get_all()))
-        self.assertEqual(5, len(services.all()))
 
     def test_service_update(self):
         old_service = self._create_service()
+        self.assertIsNotNone(old_service)
         old_updated_time = old_service.updated_at
         values = {'host': 'host-updated'}