Update db python code to bb3.8

Signed-off-by: Magnus Granberg <zorry@gentoo.org>

2 files changed, 141 insertions, 125 deletions

diff --git a/buildbot_gentoo_ci/db/connector.py b/buildbot_gentoo_ci/db/connector.py
index 7665f84..0cc7884 100644
--- a/buildbot_gentoo_ci/db/connector.py
+++ b/buildbot_gentoo_ci/db/connector.py
@@ -15,7 +15,7 @@
 # Copyright Buildbot Team Members
 # Origins: buildbot.db.connector.py
 # Modifyed by Gentoo Authors.
-# Copyright 2021 Gentoo Authors
+# Copyright 2023 Gentoo Authors
 
 import textwrap
 
@@ -70,6 +70,10 @@ class DBConnector(service.ReconfigurableServiceMixin,
         self.setName('db')
         self.basedir = basedir
 
+        # not configured yet - we don't build an engine until the first
+        # reconfig
+        self.configured_url = None
+
         # set up components
         self._engine = None  # set up in reconfigService
         self.pool = None  # set up in reconfigService
@@ -88,12 +92,16 @@ class DBConnector(service.ReconfigurableServiceMixin,
         self.builds = builds.BuildsConnectorComponent(self)
         self.workers = workers.WorkersConnectorComponent(self)
 
+        self.cleanup_timer = internet.TimerService(self.CLEANUP_PERIOD,
+                                                   self._doCleanup)
+        self.cleanup_timer.clock = self.master.reactor
+        yield self.cleanup_timer.setServiceParent(self)
+
     @defer.inlineCallbacks
     def setup(self, config, check_version=True, verbose=True):
-        db_url = config.db['db_url']
+        db_url = self.configured_url = config.db['db_url']
 
-        log.msg("Setting up database with URL %r"
-                % util.stripUrlPassword(db_url))
+        log.msg(f"Setting up database with URL {repr(util.stripUrlPassword(db_url))}")
 
         # set up the engine and pool
         self._engine = enginestrategy.create_engine(db_url,
@@ -113,3 +121,22 @@ class DBConnector(service.ReconfigurableServiceMixin,
                 for l in upgrade_message.format(basedir=self.basedir).split('\n'):
                     log.msg(l)
                 raise exceptions.DatabaseNotReadyError()
+
+    def reconfigServiceWithBuildbotConfig(self, new_config):
+        # double-check -- the master ensures this in config checks
+        assert self.configured_url == new_config.db['db_url']
+
+        return super().reconfigServiceWithBuildbotConfig(new_config)
+
+    def _doCleanup(self):
+        """
+        Perform any periodic database cleanup tasks.
+        @returns: Deferred
+        """
+        # pass on this if we're not configured yet
+        if not self.configured_url:
+            return None
+
+        d = self.changes.pruneChanges(self.master.config.changeHorizon)
+        d.addErrback(log.err, 'while pruning changes')
+        return d
diff --git a/buildbot_gentoo_ci/db/model.py b/buildbot_gentoo_ci/db/model.py
index 7ffe0ca..b80281e 100644
--- a/buildbot_gentoo_ci/db/model.py
+++ b/buildbot_gentoo_ci/db/model.py
@@ -1,7 +1,7 @@
 # This file has parts from Buildbot and is modifyed by Gentoo Authors. 
-# Buildbot is free software: you can redistribute it and/or modify it 
-# under the terms of the GNU General Public License as published by the 
-# Free Software Foundation, version 2.
+# Buildbot is free software: you can
+# redistribute it and/or modify it under the terms of the GNU General Public
+# License as published by the Free Software Foundation, version 2.
 #
 # This program is distributed in the hope that it will be useful, but WITHOUT
 # ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
@@ -18,10 +18,10 @@
 # Copyright 2023 Gentoo Authors
 
 import uuid
-import migrate
-import migrate.versioning.repository
+
+import alembic
+import alembic.config
 import sqlalchemy as sa
-from migrate import exceptions  # pylint: disable=ungrouped-imports
 
 from twisted.internet import defer
 from twisted.python import log
@@ -29,15 +29,36 @@ from twisted.python import util
 
 from buildbot.db import base
 from buildbot.db.migrate_utils import test_unicode
+from buildbot.db.types.json import JsonObject
 from buildbot.util import sautils
 
-try:
-    from migrate.versioning.schema import ControlledSchema  # pylint: disable=ungrouped-imports
-except ImportError:
-    ControlledSchema = None
+
+class UpgradeFromBefore0p9Error(Exception):
+
+    def __init__(self):
+        message = """You are trying to upgrade a buildbot 0.8.x master to buildbot 0.9.x or newer.
+        This is not supported. Please start from a clean database
+        http://docs.buildbot.net/latest/manual/upgrading/0.9-upgrade.html"""
+        # Call the base class constructor with the parameters it needs
+        super().__init__(message)
+
+
+class UpgradeFromBefore3p0Error(Exception):
+
+    def __init__(self):
+        message = """You are trying to upgrade to Buildbot 3.0 or newer from Buildbot 2.x or older.
+        This is only supported via an intermediate upgrade to newest Buildbot 2.10.x that is
+        available. Please first upgrade to 2.10.x and then try to upgrade to this version.
+        http://docs.buildbot.net/latest/manual/upgrading/3.0-upgrade.html"""
+        super().__init__(message)
 
 
 class Model(base.DBConnectorComponent):
+
+    property_name_length = 256
+    property_source_length = 256
+    hash_length = 40
+
     #
     # schema
     #
@@ -54,9 +75,6 @@ class Model(base.DBConnectorComponent):
     #
     # * dates are stored as unix timestamps (UTC-ish epoch time)
     #
-    # * sqlalchemy does not handle sa.Boolean very well on MySQL or Postgres;
-    #   use sa.SmallInteger instead
-
     # Tables related to gentoo-ci-cloud
     # -------------------------
 
@@ -452,35 +470,51 @@ class Model(base.DBConnectorComponent):
     # Migration support
     # -----------------
 
-    # this is a bit more complicated than might be expected because the first
-    # seven database versions were once implemented using a homespun migration
-    # system, and we need to support upgrading masters from that system.  The
-    # old system used a 'version' table, where SQLAlchemy-Migrate uses
-    # 'migrate_version'
+    # Buildbot has historically used 3 database migration systems:
+    #  - homegrown system that used "version" table to track versions
+    #  - SQLAlchemy-migrate that used "migrate_version" table to track versions
+    #  - alembic that uses "alembic_version" table to track versions (current)
+    # We need to detect each case and tell the user how to upgrade.
+
+    config_path = util.sibpath(__file__, "migrations/alembic.ini")
+
+    def table_exists(self, conn, table):
+        try:
+            r = conn.execute(f"select * from {table} limit 1")
+            r.close()
+            return True
+        except Exception:
+            return False
+
+    def migrate_get_version(self, conn):
+        r = conn.execute("select version from migrate_version limit 1")
+        version = r.scalar()
+        r.close()
+        return version
 
-    repo_path = util.sibpath(__file__, "migrate")
+    def alembic_get_scripts(self):
+        alembic_config = alembic.config.Config(self.config_path)
+        return alembic.script.ScriptDirectory.from_config(alembic_config)
+
+    def alembic_stamp(self, conn, alembic_scripts, revision):
+        context = alembic.runtime.migration.MigrationContext.configure(conn)
+        context.stamp(alembic_scripts, revision)
 
     @defer.inlineCallbacks
     def is_current(self):
-        if ControlledSchema is None:
-            # this should have been caught earlier by enginestrategy.py with a
-            # nicer error message
-            raise ImportError("SQLAlchemy/SQLAlchemy-Migrate version conflict")
-
-        def thd(engine):
-            # we don't even have to look at the old version table - if there's
-            # no migrate_version, then we're not up to date.
-            repo = migrate.versioning.repository.Repository(self.repo_path)
-            repo_version = repo.latest
-            try:
-                # migrate.api doesn't let us hand in an engine
-                schema = ControlledSchema(engine, self.repo_path)
-                db_version = schema.version
-            except exceptions.DatabaseNotControlledError:
+        def thd(conn):
+            if not self.table_exists(conn, 'alembic_version'):
                 return False
 
-            return db_version == repo_version
-        ret = yield self.db.pool.do_with_engine(thd)
+            alembic_scripts = self.alembic_get_scripts()
+            current_script_rev_head = alembic_scripts.get_current_head()
+
+            context = alembic.runtime.migration.MigrationContext.configure(conn)
+            current_rev = context.get_current_revision()
+
+            return current_rev == current_script_rev_head
+
+        ret = yield self.db.pool.do(thd)
         return ret
 
     # returns a Deferred that returns None
@@ -493,94 +527,49 @@ class Model(base.DBConnectorComponent):
     @defer.inlineCallbacks
     def upgrade(self):
 
-        # here, things are a little tricky.  If we have a 'version' table, then
-        # we need to version_control the database with the proper version
-        # number, drop 'version', and then upgrade.  If we have no 'version'
-        # table and no 'migrate_version' table, then we need to version_control
-        # the database.  Otherwise, we just need to upgrade it.
-
-        def table_exists(engine, tbl):
-            try:
-                r = engine.execute("select * from {} limit 1".format(tbl))
-                r.close()
-                return True
-            except Exception:
-                return False
+        # the upgrade process must run in a db thread
+        def thd(conn):
+            alembic_scripts = self.alembic_get_scripts()
+            current_script_rev_head = alembic_scripts.get_current_head()
 
-        # http://code.google.com/p/sqlalchemy-migrate/issues/detail?id=100
-        # means  we cannot use the migrate.versioning.api module.  So these
-        # methods perform similar wrapping functions to what is done by the API
-        # functions, but without disposing of the engine.
-        def upgrade(engine):
-            schema = ControlledSchema(engine, self.repo_path)
-            changeset = schema.changeset(None)
-            with sautils.withoutSqliteForeignKeys(engine):
-                for version, change in changeset:
-                    log.msg('migrating schema version {} -> {}'.format(version, version + 1))
-                    schema.runchange(version, change, 1)
-
-        def check_sqlalchemy_migrate_version():
-            # sqlalchemy-migrate started including a version number in 0.7; we
-            # support back to 0.6.1, but not 0.6.  We'll use some discovered
-            # differences between 0.6.1 and 0.6 to get that resolution.
-            version = getattr(migrate, '__version__', 'old')
-            if version == 'old':
-                try:
-                    from migrate.versioning import schemadiff
-                    if hasattr(schemadiff, 'ColDiff'):
-                        version = "0.6.1"
-                    else:
-                        version = "0.6"
-                except Exception:
-                    version = "0.0"
-            version_tup = tuple(map(int, version.split('-', 1)[0].split('.')))
-            log.msg("using SQLAlchemy-Migrate version {}".format(version))
-            if version_tup < (0, 6, 1):
-                raise RuntimeError(("You are using SQLAlchemy-Migrate {}. "
-                                    "The minimum version is 0.6.1.").format(version))
-
-        def version_control(engine, version=None):
-            ControlledSchema.create(engine, self.repo_path, version)
+            #if self.table_exists(conn, 'version'):
+            #    raise UpgradeFromBefore0p9Error()
 
-        # the upgrade process must run in a db thread
-        def thd(engine):
-            # if the migrate_version table exists, we can just let migrate
-            # take care of this process.
-            if table_exists(engine, 'migrate_version'):
-                r = engine.execute(
-                    "select version from migrate_version limit 1")
-                old_version = r.scalar()
-                if old_version < 40:
-                    raise EightUpgradeError()
-                try:
-                    upgrade(engine)
-                except sa.exc.NoSuchTableError as e:  # pragma: no cover
-                    if 'migration_tmp' in str(e):
-                        log.err('A serious error has been encountered during the upgrade. The '
-                                'previous upgrade has been likely interrupted. The database has '
-                                'been damaged and automatic recovery is impossible.')
-                        log.err('If you believe this is an error, please submit a bug to the '
-                                'Buildbot project.')
-                    raise
-
-            # if the version table exists, then we can version_control things
-            # at that version, drop the version table, and let migrate take
-            # care of the rest.
-            elif table_exists(engine, 'version'):
-                raise EightUpgradeError()
-
-            # otherwise, this db is new, so we don't bother using the migration engine
-            # and just create the tables, and put the version directly to
-            # latest
-            else:
-                # do some tests before getting started
-                test_unicode(engine)
+            if self.table_exists(conn, 'migrate_version'):
+                version = self.migrate_get_version(conn)
+
+                #if version < 40:
+                #    raise UpgradeFromBefore0p9Error()
 
+                last_sqlalchemy_migrate_version = 0
+                if version != last_sqlalchemy_migrate_version:
+                    raise UpgradeFromBefore3p0Error()
+
+                self.alembic_stamp(conn, alembic_scripts, alembic_scripts.get_base())
+                conn.execute('drop table migrate_version')
+
+            if not self.table_exists(conn, 'alembic_version'):
                 log.msg("Initializing empty database")
-                Model.metadata.create_all(engine)
-                repo = migrate.versioning.repository.Repository(self.repo_path)
 
-                version_control(engine, repo.latest)
+                # Do some tests first
+                test_unicode(conn)
+
+                Model.metadata.create_all(conn)
+                self.alembic_stamp(conn, alembic_scripts, current_script_rev_head)
+                return
+
+            context = alembic.runtime.migration.MigrationContext.configure(conn)
+            current_rev = context.get_current_revision()
+
+            if current_rev == current_script_rev_head:
+                log.msg('Upgrading database: the current database schema is already the newest')
+                return
+
+            log.msg('Upgrading database')
+            with sautils.withoutSqliteForeignKeys(conn):
+                with context.begin_transaction():
+                    context.run_migrations()
+
+            log.msg('Upgrading database: done')
 
-        check_sqlalchemy_migrate_version()
-        yield self.db.pool.do_with_engine(thd)
+        yield self.db.pool.do(thd)