molt-replicator

molt replicator

ソースデータベースから CockroachDB への継続的な変更データキャプチャ（CDC）レプリケーションです。molt fetch による初回の一括ロードが完了した後に実行します。

Important: replicator は molt とは別のバイナリです。molt fetch からは起動されません。molt fetch の data-load-and-replication モードは非推奨です。代わりに replicator を直接使用してください。

アーキテクチャ

Source DB ──► [replicator] ──► Staging DB (_replicator schema) ──► Target CockroachDB
                  ▲
            Publication/
            Slot/BinLog/
            LogMiner

Replicator はソースから変更を読み取り、ターゲット CRDB クラスター上のステージングスキーマにバッファーした後、ターゲットテーブルに適用します。

ソース別サブコマンド

ソース	コマンド
PostgreSQL	replicator pglogical
MySQL	replicator mylogical
Oracle	replicator oraclelogminer
Kafka	replicator kafka
クラウドストレージ	replicator objstore
CockroachDB CDC	replicator start

fetch から replicator までの全ワークフロー

ステップ 1：molt fetch で初回の一括ロードを実行

drop-on-target-and-recreate は既存のターゲットテーブルを最初に削除します。新規または空のターゲットに対してのみ実行してください。

molt fetch \
  --source "postgresql://<user>:<password>@source:5432/db" \
  --target "postgresql://root@crdb:26257/db" \
  --bucket-path "s3://mybucket/migration" \
  --table-handling drop-on-target-and-recreate

ステップ 2：ソースにパブリケーションを作成（PostgreSQL）

-- Run on source PostgreSQL:
CREATE PUBLICATION molt_fetch FOR ALL TABLES;
-- (molt fetch may have already created this; check first)

ステップ 3：ターゲットにステージングデータベースを作成

-- Run on target CockroachDB:
CREATE DATABASE _replicator;

ステップ 4：接続をテスト

# preflight only takes --stagingConn and --targetConn (always required for the
# target; stagingConn required if the target is not CRDB)
replicator preflight \
  --stagingConn "postgresql://root@crdb:26257/_replicator" \
  --targetConn "postgresql://root@crdb:26257/db"

ステップ 5：replicator を起動

replicator pglogical \
  --publicationName "molt_fetch" \
  --sourceConn "postgresql://<user>:<password>@source:5432/db" \
  --stagingConn "postgresql://root@crdb:26257/_replicator" \
  --stagingSchema "_replicator.public" \
  --targetConn "postgresql://root@crdb:26257/db" \
  --targetSchema "public" \
  --metricsAddr "0.0.0.0:8080"

ステップ 6：遅延を監視

curl http://localhost:8080/metrics | grep replicator_
# Watch for: mutations applied, unapplied mutations, lag

ステップ 7：切り替え

1. 遅延が約 0 になったら、アプリの書き込み先を CockroachDB に変更します
2. replicator が残りの変更を処理し終えるまで待ちます
3. ソースへの新しい書き込みがないことを確認します
4. replicator を停止します
5. molt verify が成功し、ソースのバックアップが確認されてから、ソースを廃止します

ソース固有の設定

PostgreSQL（pglogical）

ソースの前提条件：

• REPLICATION 権限を持つユーザー
• 論理レプリケーションが有効（wal_level = logical）
• パブリケーションが存在する（molt fetch によって作成、または手動で作成）

replicator pglogical \
  --publicationName "molt_fetch" \
  --slotName "replicator" \
  --sourceConn "postgresql://..." \
  --stagingConn "postgresql://root@crdb:26257/_replicator" \
  --stagingSchema "_replicator.public" \
  --targetConn "postgresql://root@crdb:26257/db" \
  --targetSchema "public"

MySQL（mylogical）

ソースの前提条件：

• バイナリログが有効（binlog_format = ROW）
• GTID モードが有効（gtid_mode=ON、enforce_gtid_consistency=ON）
• REPLICATION CLIENT 権限を持つユーザー

replicator mylogical \
  --sourceConn "mysql://<user>:<password>@source:3306/db" \
  --stagingConn "postgresql://root@crdb:26257/_replicator" \
  --stagingSchema "_replicator.public" \
  --targetConn "postgresql://root@crdb:26257/db" \
  --targetSchema "public"

Oracle（oraclelogminer）

ソースの前提条件：

• アーカイブログモードが有効
• 補足ログが有効
• LogMiner 権限が付与されている

replicator oraclelogminer \
  --sourceConn "oracle://<user>:<password>@oracle:1521/db" \
  --stagingConn "postgresql://root@crdb:26257/_replicator" \
  --stagingSchema "_replicator.public" \
  --targetConn "postgresql://root@crdb:26257/db" \
  --targetSchema "public"

主なフラグ

# Performance
--parallelism 16          # concurrent DB transactions (default: 16)
--flushSize 1000          # rows per batch (default: 1000)
--flushPeriod 1s          # flush interval (default: 1s)

# Staging connection pool
--stagingMaxPoolSize 128
--stagingIdleTime 1m
--stagingMaxLifetime 5m

# Target connection pool
--targetMaxPoolSize 128
--targetStatementCacheSize 128

# Retry
--maxRetries 10
--retryInitialBackoff 25ms
--retryMaxBackoff 2s

# Monitoring
--metricsAddr "0.0.0.0:8080"    # Prometheus metrics endpoint
--schemaRefresh 1m               # refresh schema cache (0 = disabled)

# Dead letter queue (failed rows instead of stopping)
--dlqTableName "replicator_dlq"

# Logging
-v                               # debug
-vv                              # trace
--logFormat fluent                # for log aggregators
--logDestination "/var/log/replicator.log"

注意事項

• ステージングスキーマ（_replicator.public）は replicator によって自動作成されますが、データベース（_replicator）は事前に存在している必要があります
• --publicationName と --slotName は、molt fetch が作成したものと一致させる必要があります。molt fetch の --pglogical-publication-name のデフォルトは molt_fetch で、--pglogical-replication-slot-name にはデフォルト値がありません。一方、replicator 側の --publicationName にもデフォルト値がなく、--slotName のデフォルトは replicator です。名前が一致しない場合は、両方の側で明示的に設定してください。
• DLQ テーブルは時間とともに大きくなります。失敗した行を定期的に監視して削除してください
• Replicator はソース上のレプリケーションスロットを開いたままにするため、WAL のクリーンアップが妨げられます。ソースのディスク使用量を監視してください
• 正常終了では --gracePeriod（デフォルト：30s）が適用されます。これを考慮せずに SIGKILL を送信しないでください
• アラート機能は組み込まれていません。Prometheus メトリクスエンドポイントに対する外部アラートを設定してください
• 切り替え期間が長いほどレプリケーション遅延が増加します。必要に応じてメンテナンス時間を計画してください

すべてのフラグについては、フラグリファレンスを参照してください。