DevOpsツールとベストプラクティスの習得 - DevOpsナレッジハブ DevOpsツールとベストプラクティスの習得 - DevOpsナレッジハブ

PostgreSQLにおける論理レプリケーションの実装：マルチマスターと選択的データ同期

選択的同期、アップグレード、レポートコピー、注意深くスコープされた双方向設計のためのPostgreSQL論理レプリケーションのセットアップ。

PostgreSQLにおける論理レプリケーションの実装: マルチマスターと選択的データ同期

PostgreSQLの論理レプリケーションは、クラスター全体ではなく選択したテーブルをコピーする必要がある場合に役立ちます。レポートデータベース、バージョンアップグレード、リージョナル読み取りコピー、選択的データ同期に有用ですが、ネイティブのPostgreSQL論理レプリケーションは、ターンキーで競合のないマルチマスターシステムではありません。

論理レプリケーションと物理レプリケーションの比較

物理ストリーミングレプリケーション

  • データベースクラスター全体をレプリケート
  • バイナリレベルのレプリケーション
  • 読み取り専用レプリカ
  • 同じPostgreSQLバージョンが必要
  • 低オーバーヘッド

論理レプリケーション

  • 選択的なテーブル/行レプリケーション
  • クロスバージョン互換
  • 書き込み可能なサブスクライバー
  • 高オーバーヘッド
  • 柔軟なデータ分散

論理レプリケーションのユースケース

  1. 選択的データ分散: 特定のテーブルを異なるリージョンにレプリケート
  2. 双方向同期実験: 注意深くスコープされたデータベース間の書き込み（通常はアプリケーションレベルの競合ルールあり）
  3. クロスバージョンアップグレード: 古いPostgreSQLバージョンから新しいバージョンへのレプリケーション
  4. データ集約: 複数のソースからのデータ統合
  5. GDPR準拠: 機密性の低いカラムのみをレプリケート

前提条件とセットアップ

設定要件

パブリッシャー（ソース）側:

# postgresql.conf
wal_level = logical
max_replication_slots = 10
max_wal_senders = 10

サブスクライバー（ターゲット）側:

# postgresql.conf
max_replication_slots = 10
max_logical_replication_workers = 10
max_worker_processes = 16

PostgreSQLの再起動

sudo systemctl restart postgresql

ネットワーク設定

データベースが通信できることを確認:

# サブスクライバーからパブリッシャーへの接続テスト
psql -h publisher.example.com -U replication_user -d source_db

パブリッシャーのpg_hba.confを設定:

# レプリケーション接続を許可
host    source_db    replication_user    subscriber_ip/32    scram-sha-256

基本的な論理レプリケーションのセットアップ

ステップ1: レプリケーションユーザーの作成

パブリッシャー側:

CREATE ROLE replication_user WITH REPLICATION LOGIN PASSWORD 'secure_password';
GRANT SELECT ON ALL TABLES IN SCHEMA public TO replication_user;
ALTER DEFAULT PRIVILEGES IN SCHEMA public GRANT SELECT ON TABLES TO replication_user;

ステップ2: ソーステーブルの作成

パブリッシャー側:

CREATE TABLE users (
    id SERIAL PRIMARY KEY,
    username VARCHAR(50) NOT NULL,
    email VARCHAR(100) NOT NULL,
    status VARCHAR(20) DEFAULT 'active',
    created_at TIMESTAMP DEFAULT NOW()
);

CREATE TABLE orders (
    id SERIAL PRIMARY KEY,
    user_id INTEGER REFERENCES users(id),
    amount DECIMAL(10,2),
    status VARCHAR(20),
    created_at TIMESTAMP DEFAULT NOW()
);

INSERT INTO users (username, email) VALUES 
    ('alice', '[email protected]'),
    ('bob', '[email protected]');

ステップ3: パブリケーションの作成

パブリッシャー側:

-- すべてのテーブルを公開
CREATE PUBLICATION my_publication FOR ALL TABLES;

-- または特定のテーブルを公開
CREATE PUBLICATION my_publication FOR TABLE users, orders;

-- または公開テーブルに行フィルターを適用
CREATE PUBLICATION active_users FOR TABLE users WHERE (status = 'active');

パブリケーションの表示:

SELECT * FROM pg_publication;
SELECT * FROM pg_publication_tables;

ステップ4: レプリカテーブルの作成

サブスクライバー側:

-- テーブルは同一構造である必要があります
CREATE TABLE users (
    id SERIAL PRIMARY KEY,
    username VARCHAR(50) NOT NULL,
    email VARCHAR(100) NOT NULL,
    status VARCHAR(20) DEFAULT 'active',
    created_at TIMESTAMP DEFAULT NOW()
);

CREATE TABLE orders (
    id SERIAL PRIMARY KEY,
    user_id INTEGER REFERENCES users(id),
    amount DECIMAL(10,2),
    status VARCHAR(20),
    created_at TIMESTAMP DEFAULT NOW()
);

ステップ5: サブスクリプションの作成

サブスクライバー側:

CREATE SUBSCRIPTION my_subscription
    CONNECTION 'host=publisher.example.com port=5432 dbname=source_db user=replication_user password=secure_password'
    PUBLICATION my_publication;

ステップ6: レプリケーションの確認

パブリッシャー側:

SELECT * FROM pg_stat_replication;
SELECT * FROM pg_replication_slots;

サブスクライバー側:

SELECT * FROM pg_stat_subscription;
SELECT * FROM users;  -- レプリケートされたデータが表示されるはず

高度な設定

カラムレベルのレプリケーション（PostgreSQL 15以降）

特定のカラムのみをレプリケート:

-- パブリッシャー側: 機密性の低いカラムのみをレプリケート
CREATE TABLE customers (
    id SERIAL PRIMARY KEY,
    name VARCHAR(100),
    email VARCHAR(100),
    credit_card VARCHAR(20),  -- レプリケートされない
    created_at TIMESTAMP
);

CREATE PUBLICATION customer_basic 
    FOR TABLE customers (id, name, email, created_at);

行フィルタリング

アクティブなレコードのみをレプリケート:

CREATE PUBLICATION active_data 
    FOR TABLE orders WHERE (status IN ('pending', 'processing'));

リージョナルデータ分散:

CREATE PUBLICATION us_customers 
    FOR TABLE customers WHERE (country = 'US');

CREATE PUBLICATION eu_customers 
    FOR TABLE customers WHERE (country IN ('UK', 'DE', 'FR'));

複数のパブリケーション

-- パブリッシャー: 複数のパブリケーションを作成
CREATE PUBLICATION oltp_data FOR TABLE users, orders;
CREATE PUBLICATION analytics_data FOR TABLE logs, metrics;

-- サブスクライバー: 複数のパブリケーションにサブスクライブ
CREATE SUBSCRIPTION multi_sub
    CONNECTION 'host=publisher port=5432 dbname=mydb user=repuser password=pass'
    PUBLICATION oltp_data, analytics_data;

双方向レプリケーションの注意点

ネイティブの論理レプリケーションは双方向に配線できますが、PostgreSQLは競合する書き込みを自動的にマージしません。このパターンは、各行に単一の書き込み元があり、キーが衝突せず、アプリケーションが競合を処理できる場合にのみ使用してください。

双方向同期のスケルトン例

データベースAの設定:

-- パブリケーションを作成
CREATE PUBLICATION db_a_pub FOR TABLE shared_table;

-- データベースBにサブスクライブ
CREATE SUBSCRIPTION db_a_sub
    CONNECTION 'host=db-b.example.com dbname=mydb user=repuser'
    PUBLICATION db_b_pub
    WITH (origin = none);  -- レプリケーションループを防止

データベースBの設定:

-- パブリケーションを作成
CREATE PUBLICATION db_b_pub FOR TABLE shared_table;

-- データベースAにサブスクライブ
CREATE SUBSCRIPTION db_b_sub
    CONNECTION 'host=db-a.example.com dbname=mydb user=repuser'
    PUBLICATION db_a_pub
    WITH (origin = none);

競合処理

論理レプリケーションは自動的な「最終書き込み勝ち」の競合解決を提供しません。競合する挿入、更新時の行欠落、制約違反、重複キーは、データ問題を修正するまで適用ワーカーを停止させる可能性があります。

-- 競合を追跡するためにレプリカIDを設定
ALTER TABLE shared_table REPLICA IDENTITY FULL;

競合削減戦略:

  1. タイムスタンプベースの所有権チェック: updated_atと書き込み元識別子を追加し、アプリケーションが古い書き込みを拒否できるようにします。
CREATE TABLE shared_table (
    id SERIAL PRIMARY KEY,
    data TEXT,
    updated_at TIMESTAMP DEFAULT NOW()
);

CREATE OR REPLACE FUNCTION update_timestamp()
RETURNS TRIGGER AS $$
BEGIN
    NEW.updated_at = NOW();
    RETURN NEW;
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER update_shared_table_timestamp
    BEFORE UPDATE ON shared_table
    FOR EACH ROW
    EXECUTE FUNCTION update_timestamp();
  1. バージョン番号: アプリケーションでバージョンをインクリメントし、古いバージョンに基づく更新を拒否します。
CREATE TABLE shared_table (
    id SERIAL PRIMARY KEY,
    data TEXT,
    version INTEGER DEFAULT 1
);

初期データ同期オプション

オプション1: 自動コピー（デフォルト）

-- サブスクライバーが既存データを自動的にコピー
CREATE SUBSCRIPTION my_sub
    CONNECTION 'host=publisher dbname=mydb user=repuser'
    PUBLICATION my_pub
    WITH (copy_data = true);  -- デフォルト

オプション2: 手動初期同期

大規模データセットの場合はpg_dumpを使用:

# パブリッシャーから特定のテーブルをダンプ
pg_dump -h publisher.example.com -U postgres -d source_db \
    -t users -t orders --no-owner --no-acl > initial_data.sql

# サブスクライバーにロード
psql -h subscriber.example.com -U postgres -d target_db < initial_data.sql

# 初期コピーなしでサブスクリプションを作成
psql -h subscriber.example.com -U postgres -d target_db -c "
    CREATE SUBSCRIPTION my_sub
    CONNECTION 'host=publisher dbname=source_db user=repuser'
    PUBLICATION my_pub
    WITH (copy_data = false);
"

オプション3: 大規模進行中トランザクションのストリーミング

-- コミットを待たずに大規模進行中トランザクションをストリーミング
CREATE SUBSCRIPTION fast_sync
    CONNECTION 'host=publisher dbname=mydb user=repuser'
    PUBLICATION my_pub
    WITH (
        copy_data = true,
        streaming = on,
        synchronous_commit = off
    );

論理レプリケーションの監視

パブリッシャーの監視

-- レプリケーションスロットの表示
SELECT 
    slot_name,
    plugin,
    active,
    pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn)) AS lag_size
FROM pg_replication_slots;

-- アクティブなレプリケーション接続の表示
SELECT 
    pid,
    usename,
    application_name,
    client_addr,
    state,
    sync_state,
    pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), sent_lsn)) AS send_lag
FROM pg_stat_replication;

サブスクライバーの監視

-- サブスクリプションステータスの表示
SELECT 
    subname,
    pid,
    received_lsn,
    latest_end_lsn,
    pg_size_pretty(pg_wal_lsn_diff(latest_end_lsn, received_lsn)) AS lag
FROM pg_stat_subscription;

-- レプリケーションエラーの確認
SELECT * FROM pg_stat_subscription WHERE last_msg_receipt_time < NOW() - INTERVAL '5 minutes';

監視スクリプト

#!/bin/bash
# logical-replication-monitor.sh

echo "=== パブリッシャーステータス ==="
psql -h publisher -d mydb -c "
    SELECT slot_name, active, 
           pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn)) AS lag
    FROM pg_replication_slots;"

echo ""
echo "=== サブスクライバーステータス ==="
psql -h subscriber -d mydb -c "
    SELECT subname, pid, 
           pg_size_pretty(pg_wal_lsn_diff(latest_end_lsn, received_lsn)) AS lag
    FROM pg_stat_subscription;"

トラブルシューティング

問題1: サブスクリプションがデータを受信しない

サブスクリプションステータスの確認:

SELECT subname, subenabled, subconninfo FROM pg_subscription;
SELECT subname, pid, received_lsn, latest_end_lsn FROM pg_stat_subscription;

無効な場合、サブスクリプションを有効化:

ALTER SUBSCRIPTION my_sub ENABLE;

エラーの確認:

SELECT * FROM pg_stat_subscription;

問題2: レプリケーションラグが増大

遅いテーブルの特定:

SELECT schemaname, tablename, n_tup_ins, n_tup_upd, n_tup_del
FROM pg_stat_user_tables
ORDER BY (n_tup_ins + n_tup_upd + n_tup_del) DESC;

ワーカープロセスの増加:

# サブスクライバーのpostgresql.conf
max_logical_replication_workers = 20
max_worker_processes = 30

問題3: レプリケーションスロットの肥大化

スロット使用状況の確認:

SELECT 
    slot_name,
    pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn)) AS retained_wal
FROM pg_replication_slots;

非アクティブなスロットの削除:

SELECT pg_drop_replication_slot('inactive_slot_name');

問題4: 初期同期の失敗

初期同期の再実行:

-- サブスクリプションを削除して再作成
DROP SUBSCRIPTION my_sub;

CREATE SUBSCRIPTION my_sub
    CONNECTION 'host=publisher dbname=mydb user=repuser'
    PUBLICATION my_pub
    WITH (copy_data = true);

パフォーマンス最適化

1. 適切なレプリカIDの使用

-- デフォルト: PRIMARY KEYのみ
ALTER TABLE my_table REPLICA IDENTITY DEFAULT;

-- フル: すべてのカラム（高オーバーヘッド）
ALTER TABLE my_table REPLICA IDENTITY FULL;

-- インデックス: 特定のユニークインデックスを使用
CREATE UNIQUE INDEX replica_idx ON my_table(col1, col2);
ALTER TABLE my_table REPLICA IDENTITY USING INDEX replica_idx;

2. 大規模トランザクションの並列適用

CREATE SUBSCRIPTION parallel_sub
    CONNECTION 'host=publisher dbname=mydb user=repuser'
    PUBLICATION my_pub
    WITH (streaming = parallel);

ベストプラクティス

  1. 常にレプリカIDを使用: テーブルにPRIMARY KEYまたはUNIQUE制約があることを確認
  2. レプリケーションラグを監視: リカバリ要件とWAL保持予算に基づいてアラートを設定
  3. 行フィルターを慎重に使用: フィルターはパブリッシャーで評価され、パフォーマンスに影響
  4. 競合に備える: 双方向書き込みを試みる前に競合検出を実装
  5. フェイルオーバー手順をテスト: データベース間の切り替えを練習
  6. 定期的なメンテナンス: 古いレプリケーションスロットをクリーンアップ

まとめ

単一方向の論理レプリケーションから始めてください。選択的テーブル同期、レポートコピー、多くのアップグレードワークフローに信頼性があります。双方向レプリケーションは、チェックボックスではなく高度な設計問題として扱ってください：両側に書き込む前に、行の所有権、キー生成、競合処理を定義してください。