Alembicマイグレーションを複数Podで同時実行させない方法

K8s 上の FastAPI アプリケーションで、Dockerfile の CMD に alembic upgrade head を入れて Pod 起動時にマイグレーションを実行していた。replicas=1 のうちは問題なかったが、API Pod を replicas=2 に増やした途端、デプロイ時にマイグレーションが競合した。

構成と問題

デプロイ構成は以下の通り。

API Deployment     (replicas=2)  → CMD: alembic upgrade head && uvicorn ...
Scheduler Deployment (replicas=1) → CMD: alembic upgrade head && uvicorn ...
PostgreSQL StatefulSet (replicas=1)

合計3つの Pod が起動時に alembic upgrade head を実行する。Alembic のマイグレーションには冪等性がある（alembic_version テーブルで現在のリビジョンを追跡し、既に適用済みなら何もしない）ので、通常はこれで問題なかった。

しかし 同時に 実行されると、話が変わった。

Pod A: SELECT version_num FROM alembic_version → "abc123"
Pod B: SELECT version_num FROM alembic_version → "abc123"  (同じ)
Pod A: ALTER TABLE ... (migration step 1)
Pod B: ALTER TABLE ... (migration step 1)  → ERROR: already exists

2つの Pod が同じリビジョンを読み取り、同じマイグレーションを同時に適用しようとする。CREATE INDEX や ALTER TABLE ADD COLUMN が重複実行されてエラーになった。

Alembic 自体にはロック機構がない。公式ドキュメントにも「並行実行の制御はアプリケーション側で行うこと」と書かれていた。

PostgreSQL advisory lock で直列化する

PostgreSQL の pg_advisory_lock を使って、マイグレーション実行をプロセス横断で直列化する。

alembic/env.py の run_migrations_online() を修正する。

# alembic/env.py
from sqlalchemy import engine_from_config, pool, text
from alembic import context

def run_migrations_online() -> None:
    connectable = engine_from_config(
        config.get_section(config.config_ini_section, {}),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )

    with connectable.connect() as connection:
        # PostgreSQL: advisory lock でマイグレーション直列化
        is_pg = connection.dialect.name == "postgresql"
        if is_pg:
            connection.execute(
                text("SELECT pg_advisory_lock(hashtext('alembic_migrate'))")
            )

        try:
            context.configure(
                connection=connection,
                target_metadata=target_metadata,
                compare_type=True,
            )

            with context.begin_transaction():
                context.run_migrations()
        finally:
            if is_pg:
                connection.execute(
                    text("SELECT pg_advisory_unlock(hashtext('alembic_migrate'))")
                )

実装のポイント

ロック ID の生成: hashtext('alembic_migrate') で文字列から int4 のロック ID を生成する。pg_advisory_lock は bigint を受け取るので、任意の整数でもいいが、hashtext() を使えば意味のある名前から一意な ID を得られる。

hashtext の衝突リスク: hashtext() は 32-bit 整数を返すため、異なる文字列が同じ値にハッシュ衝突する可能性がゼロではない。同一クラスタ内に複数のアプリケーションが advisory lock を使っている場合は、衝突を避けるためにアプリケーション固有の定数（例: 12345678::bigint）をハードコードするか、hashtext() の結果を定数オフセットで分離することを検討する。

dialect チェック: connection.dialect.name == "postgresql" で DB を判定する。テストで SQLite を使っている場合、advisory lock は存在しないのでスキップする。これにより pytest (SQLite in-memory) と本番 (PostgreSQL) の両方で同じ env.py が動く。

try/finally: マイグレーションが成功しても失敗しても、必ず pg_advisory_unlock を呼ぶ。ロックが残るとデッドロックの原因になる。なお、pg_advisory_lock はセッション終了時に自動解放されるため、接続が切れた場合はロックも消える。

動作の流れ

3つの Pod が同時に起動した場合:

Pod A: pg_advisory_lock(hashtext('alembic_migrate'))  → 取得成功
Pod B: pg_advisory_lock(hashtext('alembic_migrate'))  → ブロック (待機)
Pod C: pg_advisory_lock(hashtext('alembic_migrate'))  → ブロック (待機)

Pod A: run_migrations() → マイグレーション実行
Pod A: pg_advisory_unlock(...)

Pod B: ロック取得 → run_migrations() → 既に適用済み、何もしない
Pod B: pg_advisory_unlock(...)

Pod C: ロック取得 → run_migrations() → 既に適用済み、何もしない
Pod C: pg_advisory_unlock(...)

最初の1つだけがマイグレーションを実行し、残りは待機後に「適用済み」として即座に終了する。

別のアプローチとの比較

initContainer で実行する方法 もある。Deployment の initContainers にマイグレーション用コンテナを定義する方法だが、複数 Pod の initContainer も同時に実行されるため、ロックの問題は同じく発生する。

Job で実行する方法 は、デプロイ前に kubectl apply -f migration-job.yaml で単独のマイグレーション Job を走らせる。確実に1プロセスだけが実行されるが、CI/CD パイプラインが複雑になる（Job の完了待ちが必要）。

advisory lock は既存の CMD に手を加えるだけで済み、CI/CD の変更が不要という点で導入コストが最も低い。

Helm の helm.sh/hook を使う方法 も Job ベースのアプローチの一形態として広く使われている。pre-upgrade フックに Job を定義することで、Helm のアップグレード処理が始まる前にマイグレーションを1回だけ実行できる。CI/CD パイプラインを変更せずに Job アプローチを採用したい場合の選択肢として参照: Helm Chart Hooks。