先日、PostgreSQLで発生した変更をfluentdに送る方法としてLogical Decodingが使えないかなと思い、fluentdのINPUTプラグインを作ってみたのでそれの使い方をまとめておきます。

このプラグインを使うと、PostgreSQLのLogical Decoding機能(WALをデコードしてどこかに転送する機能)を使って、PostgreSQL上で起こった全ての変更をfluentd上に流すことができます。

データの流れ

(PostgreSQL)->(logical decodingのプラグイン)->(pg-logicalプラグイン)->(fluentd)

SQL経由でDB上の情報を取ってくるのとは異なり、Logical DecodingではPostgreSQLが出力したトランザクションログ（WAL）から変更情報を取ってくるので、他の処理への影響を少なく抑えることができます（トリガを仕掛けたりもいりません）。また、Logical Decoding側もプラグイン形式なっているので、好きな形式に変更結果を表すことができるので、より柔軟にデータを取得できます。今回は、WALに書かれた変更をJSON形式に変更するwal2jsonを使ってみます(※)。

(※)wal2jsonはAmazon RDS for PostgreSQLでも使えるようになりました

1. PostgreSQL側のセットアップ

1-1. PostgreSQLのインストール
PostgreSQL 9.4以降に対応しています。
最低限以下の項目を設定します。

wal_level = logicalに設定します
max_wal_sendersとmax_repilcation_slotsは1以上に設定します
pg_hba.confを修正してレプリケーション接続を許可しておきます

1-2. wal2jsonのインストール
pg_logicalプラグインではLogical Decoding機能を使うため、Logical Decodingのプラグインが必要となります。wal2jsonをダウンロードして、ビルドします。

$ git clone git@github.com:eulerto/wal2json.git
$ cd wal2json
$ make USE_PGXS=1
$ sudo MAKE USE_PGXS=1 install

1-3. PostgtreSQLを起動

2. fluentd側のセットアップ

2-1. pg_logicalプラグインをインストール

$ gem get fluent-plugin-pg-logical

2-2. fluentdの設定ファイルを書く

<source>
  @type pg_logical
  host localhost
  port 5432
  user masahiko
  dbname postgres
  slotname wal2json_slot
  create_slot true
  if_not_exists true
</source>

pluginオプションには、Logical Decodingで使用するプラグイン名を指定します。今回はwal2jsonを指定します。
fluentdの起動と同時に、必要なレプリケーションスロットも作成して欲しいので、create_slotオプションをtrueに指定します
2回目以降の起動でもレプリケーションスロットを作成しようとして失敗するので、if_not_existsオプションをtrueに指定します

3. fluentd起動

起動すると下記のログと共にWALの受信が始まります

$ fluentd -c ./fluent.conf
:
2018-02-04 22:32:22 +0900 [info]: adding source type="pg_logical"
2018-02-04 22:32:22 +0900 [info]: #0 :host=>localhost :dbname=>postgres :port=>5101 :user=>masahiko :tag=> :slotname=>wal2json_slot :plugin=>wal2json :status_interval=>10
2018-02-04 22:32:22 +0900 [info]: #0 starting fluentd worker pid=31291 ppid=31260 worker=0
2018-02-04 22:32:22 +0900 [warn]: #0 super was not called in #start: called it forcedly plugin=Fluent::PgLogicalInput
2018-02-04 22:32:22 +0900 [info]: #0 fluentd worker is now running worker=0
2018-02-04 22:32:22.351157457 +0900 fluent.info: {"worker":0,"message":"fluentd worker is now running worker=0"}

PostgreSQL上での変更を受け取ってみる

PostgreSQL上にテーブルを作成して変更すると以下のようなログが出ます。

=# CREATE TABLE hoge (c int);
=# INSERT INTO hoge VALUES(2);
=# INSERT INTO hoge VALUES(1);

2018-02-05 23:23:42.157474052 +0900 :  "{\"change\":[{\"kind\":\"insert\",\"schema\":\"public\",\"table\":\"hoge\",\"columnnames\":[\"c\"],\"columntypes\":[\"integer\"],\"columnvalues\":[2]},{\"kind\":\"insert\",\"schema\":\"public\",\"table\":\"hoge\",\"columnnames\":[\"c\"],\"columntypes\":[\"integer\"],\"columnvalues\":[1]}]}"

出力されたJSONデータは整形するとこんな感じです。これはwal2jsonの仕様です。

 {                              
     "change": [                
         {                      
             "kind": "insert",  
             "table": "hoge",   
             "schema": "public",
             "columnnames": [   
                 "c"            
             ],                 
             "columntypes": [   
                 "integer"      
             ],                 
             "columnvalues": [  
                 2              
             ]                  
         },                     
         {                      
             "kind": "insert",  
             "table": "hoge",   
             "schema": "public",
             "columnnames": [   
                 "c"            
             ],                 
             "columntypes": [   
                 "integer"      
             ],                 
             "columnvalues": [  
                 1              
             ]                  
         }                      
     ]                          
 }

違うLogical Decodingプラグインを使ってみる

wal2jsonではなく、PostgreSQLのcontribモジュールとして同梱されているtest_decodingプラグインも一緒に使ってみます。

fluentd側の設定は、以下のようにしました。

<source>
  @type pg_logical
  host localhost
  port 5432
  user masahiko
  dbname postgres
  slotname wal2json_slot
  create_slot true
  if_not_exists true
  tag pg.json
</source>

<source>
  @type pg_logical
  host localhost
  port 5432
  user masahiko
  dbname postgres
  slotname testdecode_slot
  create_slot true
  if_not_exists true
  tag pg.decode
</source>

接続先は同じですが、先程の設定に加えて、test_decodingプラグインを使うレプリケーションスロットを設定しました。
(※)レプリケーションスロットは1つの接続からしか使えないので、今回のような場合は2つのレプリケーションスロットを使用します

fluentdを再起動してみます。

2018-02-04 23:18:41 +0900 [info]: adding source type="pg_logical"
2018-02-04 23:18:41 +0900 [info]: #0 :host=>localhost :dbname=>postgres :port=>5101 :user=>masahiko :tag=> :slotname=>wal2json_slot :plugin=>wal2json :status_interval=>10
2018-02-04 23:18:41 +0900 [info]: adding source type="pg_logical"
2018-02-04 23:18:41 +0900 [info]: #0 :host=>localhost :dbname=>postgres :port=>5101 :user=>masahiko :tag=> :slotname=>testdecode_slot :plugin=>test_decoding :status_interval=>10
2018-02-04 23:18:41 +0900 [info]: #0 starting fluentd worker pid=31895 ppid=31864 worker=0
2018-02-04 23:18:41 +0900 [warn]: #0 super was not called in #start: called it forcedly plugin=Fluent::PgLogicalInput
2018-02-04 23:18:41 +0900 [warn]: #0 super was not called in #start: called it forcedly plugin=Fluent::PgLogicalInput
2018-02-04 23:18:41 +0900 [info]: #0 fluentd worker is now running worker=0
2018-02-04 23:18:41 +0900 [info]: #0 pg-logical: could not create replication slot wal2json_slot
2018-02-04 23:18:41.477931908 +0900 fluent.info: {"worker":0,"message":"fluentd worker is now running worker=0"}
2018-02-04 23:18:41.502037085 +0900 fluent.info: {"message":"pg-logical: could not create replication slot wal2json_slot"}

再度、PostgreSQL上のテーブルを変更します。

=# INSERT INTO hoge SELECT generate_series(1,10);

fluentd側では以下のデータを受け取ります。

2018-02-04 15:18:55.184469399 +0900 : "BEGIN 189044"
2018-02-04 15:18:55.184852262 +0900 : "{\"change\":[{\"kind\":\"insert\",\"schema\":\"public\",\"table\":\"hoge\",\"columnnames\":[\"c\"],\"columntypes\":[\"integer\"],\"columnvalues\":[1]},{\"kind\":\"insert\",\"schema\":\"public\",\"table\":\"hoge\",\"columnnames\":[\"c\"],\"columntypes\":[\"integer\"],\"columnvalues\":[2]},{\"kind\":\"insert\",\"schema\":\"public\",\"table\":\"hoge\",\"columnnames\":[\"c\"],\"columntypes\":[\"integer\"],\"columnvalues\":[3]},{\"kind\":\"insert\",\"schema\":\"public\",\"table\":\"hoge\",\"columnnames\":[\"c\"],\"columntypes\":[\"integer\"],\"columnvalues\":[4]},{\"kind\":\"insert\",\"schema\":\"public\",\"table\":\"hoge\",\"columnnames\":[\"c\"],\"columntypes\":[\"integer\"],\"columnvalues\":[5]},{\"kind\":\"insert\",\"schema\":\"public\",\"table\":\"hoge\",\"columnnames\":[\"c\"],\"columntypes\":[\"integer\"],\"columnvalues\":[6]},{\"kind\":\"insert\",\"schema\":\"public\",\"table\":\"hoge\",\"columnnames\":[\"c\"],\"columntypes\":[\"integer\"],\"columnvalues\":[7]},{\"kind\":\"insert\",\"schema\":\"public\",\"table\":\"hoge\",\"columnnames\":[\"c\"],\"columntypes\":[\"integer\"],\"columnvalues\":[8]},{\"kind\":\"insert\",\"schema\":\"public\",\"table\":\"hoge\",\"columnnames\":[\"c\"],\"columntypes\":[\"integer\"],\"columnvalues\":[9]},{\"kind\":\"insert\",\"schema\":\"public\",\"table\":\"hoge\",\"columnnames\":[\"c\"],\"columntypes\":[\"integer\"],\"columnvalues\":[10]}]}"
2018-02-04 15:18:55.184961889 +0900 : "table public.hoge: INSERT: c[integer]:1"
2018-02-04 15:18:55.185093425 +0900 : "table public.hoge: INSERT: c[integer]:2"
2018-02-04 15:18:55.185275368 +0900 : "table public.hoge: INSERT: c[integer]:3"
2018-02-04 15:18:55.185367719 +0900 : "table public.hoge: INSERT: c[integer]:4"
2018-02-04 15:18:55.185437581 +0900 : "table public.hoge: INSERT: c[integer]:5"
2018-02-04 15:18:55.185554763 +0900 : "table public.hoge: INSERT: c[integer]:6"
2018-02-04 15:18:55.185648607 +0900 : "table public.hoge: INSERT: c[integer]:7"
2018-02-04 15:18:55.185725397 +0900 : "table public.hoge: INSERT: c[integer]:8"
2018-02-04 15:18:55.185819775 +0900 : "table public.hoge: INSERT: c[integer]:9"
2018-02-04 15:18:55.185897049 +0900 : "table public.hoge: INSERT: c[integer]:10"
2018-02-04 15:18:55.186005774 +0900 : "COMMIT 189044"

上記のように、異なる形式でPostgreSQL上の変更が転送されてきたことがわかります。