Change Data Capture

データベースから行レベルの変更をストリーミングします。CDCソースはinsert、update、deleteをキャプチャし、任意で各購読者にまず既存行の一貫したスナップショットを渡し、すべてをドライバー中立の変更イベントとして配信します。ソースはエントリIDでアドレス指定でき、Luaからはcdcモジュール経由で利用します。

エントリ種別

どちらの種別も、同じLua API、同じソース情報レコード、同じ変更イベントの形を公開します。異なるのは保証の集合であり、これはソースごとにケイパビリティとして公開されます。

Postgresの設定

- name: pg_cdc
  kind: db.cdc.postgres
  host: ${env:DB_HOST}
  port: 5432
  database: app
  username: ${env:DB_USER}
  password: ${env:app.secrets:db_password}
  slot_name: wippy_slot
  publication: wippy_pub
  tables:
    - public.users
    - public.orders
  snapshot: true
  streaming: true
  standby_interval: "10s"
  status_interval: "10s"
  lifecycle:
    auto_start: true

いずれかのmax_*フィールドに0を指定するとデフォルトが選択されます。デコーダーが無制限になることはありません。負の値は拒否されます。

認証情報の${env:NAME}プレースホルダーは、デコード時に環境レジストリを通じて解決されます。

SQLiteの設定

SQLiteソースは自前のデータベースを開きません。既存のdb.sql.sqliteリソースを借用し、そのリソースのコミット済みミューテーションのオブザーバーを購読します。したがって、そのWippy SQLリソースを通じて行われた書き込みのみを正確にキャプチャします。別のプロセス、別のコネクション、外部ツールによる書き込みは観測されません。

- name: cdcdb
  kind: db.sql.sqlite
  file: /var/data/app.db
  lifecycle:
    auto_start: true

- name: changes
  kind: db.cdc.sqlite
  db_resource: app:cdcdb
  tables:
    - users
    - orders
  snapshot: true
  status_interval: "30s"
  lifecycle:
    auto_start: true

ソースはSQLリソースをライフサイクル要件として宣言するため、スーパーバイザーは先にデータベースを起動し、データベースの世代が置き換えられるとソースを再起動します。

購読の制限

各ソースは限られた数の購読者を受け入れ、その最悪ケースのバックログを事前に予約します。スナップショットのスロットは、スナップショットを有効にしたストリームが閉じるまで予約されたままです。

subscriptions:
  max_subscriptions: 1024
  max_snapshot_subscriptions: 4
  max_bytes: 268435456

0を指定するとデフォルトが選択され、負の値は拒否されます。制限を使い切ると、購読は再試行可能なerrors.UNAVAILABLEで失敗します。

仕組み

1. Postgresソースはレプリケーションユーザーとして接続し、slot_nameで指定されたスロットを作成（または再開）します。SQLiteソースはdb_resourceを借用し、そのリソースのコミット済みミューテーションのオブザーバーを購読します。
2. 行の変更は、opがinsert、update、delete、truncateのドライバー中立な変更イベントへデコードされます。
3. ストリームでsnapshotが有効な購読者は（エントリのsnapshotフィールド、またはストリームのopts.snapshotにより）、まず既存行をop = "snapshot"のイベントとして受け取り、その後、両者の間に切れ目なくライブの変更へ続きます。
4. Postgresソースは定期的にLSNを確認応答し、サーバーがWALセグメントを解放できるようにします（standby_interval）。
5. ソースは自身のエントリIDで登録されます。Luaコードはcdc.streamで購読します。

ケイパビリティ

すべてのソースは自身が保証する内容を公開するため、コンシューマーはエントリ種別ではなくケイパビリティで分岐します。

ケイパビリティのフラグはソースの進捗を表すものであり、永続的な配信を表すものではありません。遅れをとった、あるいは切断した個々の購読者に対してイベントを再生するドライバーはありません。

ソース情報

各ソースは情報レコードで記述され、cdc.sourceとcdc.list_sourcesが返します。

admissionはキューの充填量ではなく予約を数えます。activeは受け入れられた購読数、snapshotsはそのうちスナップショットを有効にした数、reserved_bytesは予約されたバックログ予算、rejectedは制限によって拒否された購読の累計数です。

権限

CDCの権限はデータベースへのアクセスとは別です。ソースは、変更前イメージを含め、キャプチャしたすべての行を公開しうるからです。ストリームのフィルタは配信を絞り込むだけであり、ソースへのアクセスを付与することは決してありません。

- name: cdc_reader
  kind: security.policy
  policy:
    effect: allow
    actions: [cdc.source, cdc.subscribe]
    resources: [app:changes]

関連項目

• CDCモジュール - LuaのストリーミングAPI
• データベース - SQLデータベースサービス
• 環境 - ${env:NAME}による認証情報の解決
• セキュリティ - ポリシーとアクション