update: Parallel DB locks watcher

.flake8

@@ -1,5 +1,5 @@
 [flake8]
-max-line-length = 80
+max-line-length = 88
 max-complexity = 16
 # B = bugbear
 # B9 = bugbear opinionated (incl line length)

README.rst

@@ -183,14 +183,31 @@ click-odoo-update (beta)
     update based on a hash of their file content, compared to the hashes
     stored in the database.
 
+    It allows updating in parallel while another Odoo instance is still
+    running against the same database, by using a watcher that aborts the
+    update in case a DB lock happens.
+
   Options:
-    -c, --config FILE    ...
-    -d, --database TEXT  ...
-    ...
-    --i18n-overwrite     Overwrite existing translations
-    --update-all         Force a complete upgrade (-u base)
-    --if-exists          Don't report error if database doesn't exist
-    --help               Show this message and exit.
+    -c, --config FILE            Specify the Odoo configuration file. Other ways
+                                 to provide it are with the ODOO_RC or
+                                 OPENERP_SERVER environment variables, or
+                                 ~/.odoorc (Odoo >= 10) or ~/.openerp_serverrc.
+    --addons-path TEXT           Specify the addons path. If present, this
+                                 parameter takes precedence over the addons path
+                                 provided in the Odoo configuration file.
+    -d, --database TEXT          Specify the database name. If present, this
+                                 parameter takes precedence over the database
+                                 provided in the Odoo configuration file.
+    --log-level TEXT             Specify the logging level. Accepted values
+                                 depend on the Odoo version, and include debug,
+                                 info, warn, error.  [default: info]
+    --logfile FILE               Specify the log file.
+    --i18n-overwrite             Overwrite existing translations
+    --update-all                 Force a complete upgrade (-u base)
+    --if-exists                  Don't report error if database doesn't exist
+    --watcher-max-seconds FLOAT  Max DB lock seconds allowed before aborting the
+                                 update process. Default: 0 (disabled).
+    --help                       Show this message and exit.
 
 click-odoo-upgrade (deprecated, see click-odoo-update)
 ------------------------------------------------------

click_odoo_contrib/update.py

@@ -5,10 +5,14 @@
 import json
 import logging
 import os
-from contextlib import contextmanager
+import threading
+from contextlib import closing, contextmanager
+from datetime import timedelta
+from time import sleep
 
 import click
 import click_odoo
+import psycopg2
 from click_odoo import OdooEnvironment, odoo
 
 from ._addon_hash import addon_hash
@@ -21,6 +25,104 @@
 DEFAULT_EXCLUDE_PATTERNS = "*.pyc,*.pyo,i18n/*.pot,i18n_extra/*.pot,static/*"
 
 
+class DbLockWatcher(threading.Thread):
+    def __init__(self, database, max_seconds):
+        super(DbLockWatcher, self).__init__()
+        self.daemon = True
+        self.database = database
+        self.max_seconds = max_seconds
+        self.aborted = False
+        self.watching = False
+
+    def stop(self):
+        self.watching = False
+
+    def run(self):
+        """Watch DB while another process is updating Odoo.
+
+        This method will query :param:`database` to see if there are DB locks.
+        If a lock longer than :param:`max_seconds` is detected, it will be
+        terminated and an exception will be raised.
+
+        :param str database:
+            Name of the database that is being updated in parallel.
+
+        :param float max_seconds:
+            Max length of DB lock allowed.
+        """
+        _logger.info("Starting DB lock watcher")
+        beat = self.max_seconds / 3
+        max_td = timedelta(seconds=self.max_seconds)
+        own_pid_query = "SELECT pg_backend_pid()"
+        # SQL explained in https://wiki.postgresql.org/wiki/Lock_Monitoring
+        locks_query = """
+            SELECT
+                pg_stat_activity.datname,
+                pg_class.relname,
+                pg_locks.transactionid,
+                pg_locks.mode,
+                pg_locks.granted,
+                pg_stat_activity.usename,
+                pg_stat_activity.query,
+                pg_stat_activity.query_start,
+                AGE(NOW(), pg_stat_activity.query_start) AS age,
+                pg_stat_activity.pid
+            FROM
+                pg_stat_activity
+                JOIN pg_locks ON pg_locks.pid = pg_stat_activity.pid
+                JOIN pg_class ON pg_class.oid = pg_locks.relation
+            WHERE
+                NOT pg_locks.granted
+                AND pg_stat_activity.datname = %s
+            ORDER BY pg_stat_activity.query_start
+        """
+        # See https://stackoverflow.com/a/35319598/1468388
+        terminate_session = "SELECT pg_terminate_backend(%s)"
+        if odoo.release.version_info < (9, 0):
+            params = {"dsn": odoo.sql_db.dsn(self.database)[1]}
+        else:
+            params = odoo.sql_db.connection_info_for(self.database)[1]
+        # Need a separate raw psycopg2 cursor without transactioning to avoid
+        # weird concurrency errors; this cursor will only trigger SELECTs, and
+        # it needs to access current Postgres server status, monitoring other
+        # transactions' status, so running inside a normal, transactioned,
+        # Odoo cursor would block such monitoring and, after all, offer no
+        # additional protection
+        with closing(psycopg2.connect(**params)) as watcher_conn:
+            watcher_conn.set_isolation_level(
+                psycopg2.extensions.ISOLATION_LEVEL_AUTOCOMMIT
+            )
+            self.watching = True
+            while self.watching:
+                # Wait some time before checking locks
+                sleep(beat)
+                # Ensure no long blocking queries happen
+                with closing(
+                    watcher_conn.cursor(cursor_factory=psycopg2.extras.DictCursor)
+                ) as watcher_cr:
+                    if _logger.level <= logging.DEBUG:
+                        watcher_cr.execute(own_pid_query)
+                        watcher_pid = watcher_cr.fetchone()[0]
+                        _logger.debug(
+                            "DB lock watcher running with postgres PID %d", watcher_pid
+                        )
+                    watcher_cr.execute(locks_query, (self.database,))
+                    locks = watcher_cr.fetchall()
+                    if locks:
+                        _logger.warning("%d locked queries found", len(locks))
+                        _logger.info("Query details: %r", locks)
+                    for row in locks:
+                        if row["age"] > max_td:
+                            # Terminate the query to abort the parallel update
+                            _logger.error(
+                                "Long locked query detected; aborting update cursor "
+                                "with PID %d...",
+                                row["pid"],
+                            )
+                            self.aborted = True
+                            watcher_cr.execute(terminate_session, (row["pid"],))
+
+
 def _get_param(cr, key, default=None):
     cr.execute("SELECT value FROM ir_config_parameter WHERE key=%s", (key,))
     r = cr.fetchone()
@@ -52,6 +154,7 @@ def _save_installed_checksums(cr):
     for (module_name,) in cr.fetchall():
         checksums[module_name] = _get_checksum_dir(cr, module_name)
     _set_param(cr, PARAM_INSTALLED_CHECKSUMS, json.dumps(checksums))
+    _logger.info("Database updated, new checksums stored")
 
 
 def _get_checksum_dir(cr, module_name):
@@ -69,11 +172,10 @@ def _get_checksum_dir(cr, module_name):
     return checksum_dir
 
 
-@contextmanager
-def OdooEnvironmentWithUpdate(database, ctx, **kwargs):
+def _update_db(database, update_all, i18n_overwrite, watcher=None):
     conn = odoo.sql_db.db_connect(database)
     to_update = odoo.tools.config["update"]
-    if ctx.params["update_all"]:
+    if update_all:
         to_update["base"] = 1
     else:
         with conn.cursor() as cr:
@@ -87,15 +189,40 @@ def OdooEnvironmentWithUpdate(database, ctx, **kwargs):
                 "Updating addons for their hash changed: %s.",
                 ",".join(to_update.keys()),
             )
-    if ctx.params["i18n_overwrite"]:
+    if i18n_overwrite:
         odoo.tools.config["overwrite_existing_translations"] = True
     if odoo.release.version_info[0] < 10:
         Registry = odoo.modules.registry.RegistryManager
     else:
         Registry = odoo.modules.registry.Registry
     Registry.new(database, update_module=True)
+    if watcher and watcher.aborted:
+        # If you get here, the updating session has been terminated and it
+        # somehow has recovered by opening a new cursor and continuing;
+        # that's very unlikely, but just in case some paranoid module
+        # happens to update, let's just make sure the exit code for
+        # this script indicates always a failure
+        raise click.Abort("Update aborted by watcher, check logs")
     with conn.cursor() as cr:
         _save_installed_checksums(cr)
+
+
+@contextmanager
+def OdooEnvironmentWithUpdate(database, ctx, **kwargs):
+    # Watch for database locks while Odoo updates
+    watcher = None
+    if ctx.params["watcher_max_seconds"] > 0:
+        watcher = DbLockWatcher(database, ctx.params["watcher_max_seconds"])
+        watcher.start()
+    # Update Odoo datatabase
+    try:
+        _update_db(
+            database, ctx.params["update_all"], ctx.params["i18n_overwrite"], watcher
+        )
+    finally:
+        if watcher:
+            watcher.stop()
+    # If we get here, the database has been updated
     with OdooEnvironment(database) as env:
         yield env
 
@@ -112,10 +239,26 @@ def OdooEnvironmentWithUpdate(database, ctx, **kwargs):
 @click.option(
     "--if-exists", is_flag=True, help="Don't report error if database doesn't exist"
 )
-def main(env, i18n_overwrite, update_all, if_exists):
+@click.option(
+    "--watcher-max-seconds",
+    default=0,
+    type=float,
+    help="Max DB lock seconds allowed before aborting the update process. "
+    "Default: 0 (disabled).",
+)
+def main(env, i18n_overwrite, update_all, if_exists, watcher_max_seconds):
     """ Update an Odoo database (odoo -u), automatically detecting
     addons to update based on a hash of their file content, compared
     to the hashes stored in the database.
+
+    If you want to update in parallel while another Odoo instance is still
+    running against the same database, you can use `--watcher-max-seconds`
+    to start a watcher thread that aborts the update in case a DB
+    lock is found. You will probably need to have at least 2 odoo codebases
+    running in parallel (the old one, serving; the new one, updating) and
+    swap them ASAP once the update is done. This process will reduce downtime
+    a lot, but it requires deeper knowledge of Odoo internals to be used
+    safely, so use it at your own risk.
     """
     if not env:
         msg = "Database does not exist"

tests/test_update.py

@@ -123,3 +123,18 @@ def test_update_i18n_overwrite(odoodb):
     ]
     subprocess.check_call(cmd)
     # TODO how to test i18n-overwrite was effectively applied?
+
+
+def test_parallel_watcher(odoodb):
+    # Test that the parallel updater does not disturb normal operation
+    cmd = [
+        sys.executable,
+        "-m",
+        "click_odoo_contrib.update",
+        "--watcher-max-seconds",
+        "30",
+        "-d",
+        odoodb,
+    ]
+    subprocess.check_call(cmd)
+    # TODO Test an actual lock