ci: SRE-477: implement migrate lock (HumanSignal#3977)

deploy/docker-entrypoint.d/common/30-run-db-migrations.sh

@@ -3,7 +3,7 @@ set -e ${DEBUG:+-x}
 
 if [ -n "${POSTGRE_HOST:-}" ] || [ -n "${MYSQL_HOST:-}" ] && [ "${SKIP_DB_MIGRATIONS:-}" != "true" ]; then
   echo >&3 "=> Do database migrations..."
-  python3 /label-studio/label_studio/manage.py migrate >&3
+  python3 /label-studio/label_studio/manage.py locked_migrate >&3
   echo >&3 "=> Migrations completed."
 else
   echo >&3 "=> Skipping run db migrations."

deploy/docker-entrypoint.sh

@@ -68,7 +68,7 @@ elif [ "$1" = "label-studio-uwsgi" ]; then
   exec_or_wrap_n_exec uwsgi --ini /label-studio/deploy/uwsgi.ini
 elif [ "$1" = "label-studio-migrate" ]; then
   exec_entrypoint "$ENTRYPOINT_PATH/app-init/"
-  exec python3 /label-studio/label_studio/manage.py migrate >&3
+  exec python3 /label-studio/label_studio/manage.py locked_migrate >&3
 else
   exec_or_wrap_n_exec "$@"
 fi

label_studio/core/management/commands/locked_migrate.py

@@ -0,0 +1,41 @@
+import logging
+from django.db import connections
+from django.conf import settings
+from django.core.management.commands.migrate import Command as MigrateCommand
+
+logger = logging.getLogger(__name__)
+
+DEFAULT_LOCK_ID = getattr(settings, "MIGRATE_LOCK_ID", 1000)
+
+
+class Command(MigrateCommand):
+    help = "Run Django migrations safely, using a lock"
+
+    def add_arguments(self, parser):
+        MigrateCommand.add_arguments(self, parser)
+        parser.add_argument(
+            "--migrate-lock-id",
+            default=DEFAULT_LOCK_ID,
+            type=int,
+            help="The id of the advisory lock to use",
+        )
+
+    def handle(self, *args, **options):
+        database = options["database"]
+        if not options["skip_checks"]:
+            self.check(databases=[database])
+
+        # Get the database we're operating from
+        connection = connections[database]
+        # Hook for backends needing any database preparation
+        connection.prepare_database()
+
+        lock_id = options["migrate_lock_id"]
+        with connection.cursor() as cursor:
+            try:
+                logger.debug(f"Trying to acquire lock: {lock_id}")
+                cursor.execute(f"SELECT pg_advisory_xact_lock({lock_id})")
+                logger.debug(f"Successfully acquired lock. Starting migrations...")
+                MigrateCommand.handle(self, *args, **options)
+            finally:
+                logger.debug(f"Lock released.")