Kafka、MongoDB、Maxwell's Daemonを使用したSQLデータベース監査システムの構築

原文(投稿日：2021/01/01)へのリンク

OK、つまりあなたは、トランザクションデータの保管にリレーショナルデータベースを使っていて、一部のテーブル内の特定データについて監査証跡を取得する必要がある、ということですね。あなたが多くの開発者たちと同じであれば、おそらく次のような対処をすることでしょう。

1. データベースの監査ログ機能を使用する

ほとんどのデータベースには、監査ログをサポートするプラグインが提供されているので、これらのプラグインをインストールすれば、簡単な設定でログデータを収集することができます。ただしこれには、次のような問題があります。

• 本格的な監査ログプラグインは、一般的にはエンタープライズエディションでのみ使用可能です。コミュニケーションエディションには用意されていない場合があります。例えばMySQLでは、監査ログプラグインはエンタープライズエディションでしか使用できません。ただし、コミュニティエディションのユーザであっても、MariaDBやPerconaの他の監査ログプラグインをインストールすれば、この制限を回避することが可能になります。
• こちらやこちらで論じられているように、DBレベルの監査ログは、データベースサーバに10～20パーセントのオーバーヘッドを発生させます。一般的に監査ログは、低速クエリにおいてのみ使用されるものであって、高負荷システムのすべてのクエリを対象とすることは望まれません。
• 監査ログはログファイルに記録されるため、データの検索が容易ではありません。データ分析や監査目的においては、監査データは検索可能なフォーマットであることが望まれます。
• 監査データはDBと同じサーバに保管されるため、大規模な監査アーカイブは相当量のデータベースストレージを消費します。

2. アプリケーションを使って監査ログを処理する

次のような方法が考えられます。

a. 既存のデータを更新する前に、データを別のテーブルにコピーした上で、現在のテーブルのデータを更新する。

b. データにバージョン番号を付加して、データの更新毎にインクリメントしたバージョン番号を挿入する。

c. 2つのDBテーブルに書き込みを行い、ひとつは最新データ、もうひとつは監査証跡とする。

スケーラブルなシステム設計の原則として、同一データの複数書き込みは回避するべきです。アプリケーションのパフォーマンスを低下させるだけでなく、データの非同期に関するさまざまな問題を発生させるからです。

監査データはなぜ必要か？

監査ログシステムのアーキテクチャに関する議論を進める前に、さまざまな組織における監査ログシステムの必要性について、いくつかのポイントから見ていきましょう。

• コンプライアンスと監査: 監査人には、彼らの観点から意味のある、文脈的な形式のデータが必要です。DB監査ログはDBAチーム向けのものであって、監査人のためのものではありません。
• セキュリティ違反に対して重大な警報を発する機能は、大規模なソフトウェアにおいては基本的な要件です。監査ログはこの目的に使用することが可能です。
• 誰がデータにアクセスしたのか、データの最初の状態はどうだったのか、更新された時に何が変わったのか、内部ユーザによる権限の悪用があったのかなど、さまざまな質問に答えられなくてはなりません。
• 監査証跡は侵入者の特定に有効であることから、"内部関係者"への抑止力になる点にも注目すべきです。自分の行動が精査されていると知れば、許可されていないデータにアクセスしたり、特定データを改ざんしたりする可能性は低くなります。
• 金融やエネルギから外食産業や公共事業に至るまで、あらゆる種類の産業に、データアクセスを解析して詳細なレポートを作成し、政府関係機関に定期的に報告する義務が課せられています。HIPAA(医療保険の相互運用性と説明責任に関する法律)を例にしてみましょう。HPAAでは医療機関に対して、内部記録を操作するすべての者に関する監査証跡の提供を求めています。この対象はデータ行とレコードです。EUの新しいGDPR(一般データ保護規則)にも同様の要件があります。例えばSOX(サーベインス・オクスレー法)では、公共企業に対して幅広く会計規制を課しています。対象となる組織はデータアクセスを分析して、詳細な報告書を定期的に提出しなければなりません。

この記事では、Maxwell's DaemonやKafkaのようなテクノロジを用いて、監査証跡データの管理を行うスケーラブルなソリューションを紹介したいと思います。

課題

アプリケーションやデータモデルに依存しない監査システムを構築する。ただしそのシステムは、スケーラブルでコスト効率のよいものでなくてはならない。

アーキテクチャ

重要な注意: このシステムは、MySQL DBをRowベースのbinlogログフォーマットで使用する場合にのみ機能します。

ソリューションアーキテクチャを詳細に論じる前に、この記事で取り上げるテクノロジについて、それぞれを簡単に見ていきましょう。

Maxwell's Daemon

Maxwell's Daemon (MD)はZendeskの提供するオープンソースのプロジェクトです。MySQL binlogを入力として、KafkaやKinesis、その他のストリーミングプラットフォームに行アップデートを書き込みます。Maxwellは運用オーバーヘッドが低く、こちらに説明されているようにMySQLと書き込み場所以外のものを必要としません。要するに、MDは変更データキャプチャ(CDC)ツールなのです。

RedhatのDebezium、NetflixのDBLog、LinkedInのBooklynなど、市場にはたくさんのCDCがあります。今回のセットアップは、これらツールのどれでも実現が可能ですが、NetflixのDBLogとLinkedInのBrooklynは上のリンクで説明したものとは異なるユースケースを想定して開発されています。ただしDebeziumについては、MDに極めて近いため、私たちのアーキテクチャにおいてMDの代用として使うことができます。MDとDebeziumのどちらかを選択する前に考慮すべき点について、簡単にまとめておきましょう。

• Debeziumがデータを書き込めるのはKafkaのみです — 少なくとも、サポートする主要プロダクトがKafkaである、とは言えます。一方のMDは、Kafkaを含むさまざまなプロダクトをサポートしています。MDがサポートするプロデューサはKafka、Kinesis、Google Cloud Pub/Sub、SQS、Rabbit MQ、Redisなどです。
• MDでは、自分自身でプロデューサを記述して、それを設定することができます。詳細についてはこちらに説明されています。
• Debeziumのアドバンテージは、MySQLやMongoDB、PostgreSQL、SQL Server、Cassandra、DB2、Oracleなど、複数のソースからChange Dataを読み込めることです。新たなデータソースの追加にも積極的です。一方のMDは、現時点でデータソースとしてサポートするのはMySQLのみです。

Kafka

Apache Kafkaは、オープンソースの分散イベントストリーミングプラットフォームです。ハイパフォーマンスなデータパイプラインやストリーミング解析、データ統合、ミッションクリティカルなアプリケーションに使用されています。

MongoDB

MongoDBは、最新のアプリケーション開発者とクラウド時代のために開発された、汎用でドキュメントベースの分散データベースです。ここでは説明のためにMongoDBを選択しましたが、S3など他のオプションや、influxDBやCassandraといった時系列データベースを選択することも可能です。

下の図は、監査証跡ソリューションのためのデータフローダイアグラムを示すものです。

図１. データフローダイアグラム

監査証跡管理システムに関連する手順は次のとおりです。

1. アプリケーションが書き込み、更新、削除などのDB操作を実行する。
2. SQL DBがオペレーションのバイナリログをROWフォーマットで生成する。これはSQL DBのコンフィギュレーションで指定する。
3. Maxwell's DaemonがSQLバイナリログをポーリングして、新たなエントリを読み、Kafka Topicにそれを書き込む。
4. コンシューマアプリがKafka Topicをポーリングしてデータを読み込み、処理する。
5. 処理されたデータがコンシューマによって、新たなデータストアに書き込まれる。

セットアップ

セットアップを簡単にするため、可能な場所ではDockerコンテナを使用します。Dockerがマシンにインストールされていなければ、Docker Desktopのインストールを検討してください。

MySQL DB

1. mysqlサーバをローカルで実行する。以下のコマンドは、mysqlコンテナをポート3307上で起動するものです。

docker run -p 3307:3306 -p 33061:33060 --name=mysql83 -d mysql/mysql-server:latest

2.    最初のインストールでrootのパスワードが分からなければ、次のコマンドを実行してパスワードをコンソールに表示させます。

docker logs mysql83 2>&1 | grep GENERATED

3. コンテナにログインして、必要ならばパスワードを変更します。

docker exec -it mysql83 mysql -uroot -p
alter user 'root'@'localhost' IDENTIFIED BY 'abcd1234'

4. セキュリティ上の理由から、dockerコンテナは、デフォルトではアプリケーション外部からの接続が許されていないので、次のコマンドを実行して変更する必要があります。

update mysql.user set host = '%' where user='root';

5. mysqlプロンプトを抜けてdockerコンテナを再起動します。

docker container restart mysql83

6. 再度mysqlクライアントにログインし、次のコマンドを実行して、maxwell's daemon用のユーザを作ります。このステップの詳細は"Maxwell's Daemon Quick Start"を参照してください。

docker exec -it mysql83 mysql -uroot -p
set global binlog_format=ROW;
set global binlog_row_image=FULL;
CREATE USER 'maxwell'@'%' IDENTIFIED BY 'pmaxwell';
GRANT ALL ON maxwell.* TO 'maxwell'@'%';
GRANT SELECT, REPLICATION CLIENT, REPLICATION SLAVE ON *.* TO 'maxwell'@'%';
CREATE USER 'maxwell'@'localhost' IDENTIFIED BY 'pmaxwell';
GRANT ALL ON maxwell.* TO 'maxwell'@'localhost';
GRANT SELECT, REPLICATION CLIENT, REPLICATION SLAVE ON *.* TO 'maxwell'@'localhost';

Kafkaブローカ

Kafkaのセットアップは非常に簡単です。このリンクからダウンロードして、

以降のコマンドを実行してください。

まず、Kafkaを取り出します。

tar -xzf kafka_2.13-2.6.0.tgz
cd kafka_2.13-2.6.0

Zookeeperを起動します。Kafkaを使用するために、現時点ではこれが必要です。

bin/zookeeper-server-start.sh config/zookeeper.properties

別のターミナルでKafkaを起動します。

bin/kafka-server-start.sh config/server.properties

別のターミナルでトピックを生成します。

bin/kafka-topics.sh --create --topic maxwell-events --bootstrap-server localhost:9092 --partitions 1 --replication-factor 1

以上のコマンドでKafkaブローカが起動して、その中に"maxwell-events"トピックが生成されます。

メッセージをKafkaトピックにパブリッシュするには、新しいターミナルで次のコマンドを実行してください。

bin/kafka-console-producer.sh --topic maxwell-events --broker-list localhost:9092

プロンプトが表示されるので、メッセージを入力してエンターを押せば、そのメッセージがKafkaに送信されます。

Kafkaトピックからメッセージをコンシュームします。

bin/kafka-console-producer.sh --topic quickstart-events --broker-list localhost:9092

Maxwell's Daemon

maxwell's daemonをここからダウンロードしてください。
tarファイルを解凍して、次のコマンドを実行します。

bin/maxwell --user=maxwell   --password=pmaxwell --host=localhost --port=3307  --producer=kafka     --kafka.bootstrap.servers=localhost:9092 --kafka_topic=maxwell-events

これによって、先ほど説明したデータベースセットアップのバイナリログを監視するように、Maxwellがセットアップされます。データベース内の特定のDBやテーブルのみを監視することも、もちろん可能です。詳細な情報は"Maxwell's Daemon Configuration"の資料を参照してください。

セットアップをテストする

セットアップが正しく動作するかどうかをテストするために、MySQLに接続して、テーブルにいくつかのデータを挿入してみます。

docker exec -it mysql83 mysql -uroot -p

CREATE DATABASE maxwelltest;

USE maxwelltest;

CREATE TABLE Persons (
    PersonId int NOT NULL AUTO_INCREMENT,
    LastName varchar(255),
    FirstName varchar(255),
    City varchar(255),
    primary key (PersonId)

);

INSERT INTO Persons (LastName, FirstName, City) VALUES ('Erichsen', 'Tom',  'Stavanger');

次に、別のターミナルから、以下のコマンドを実行します。

bin/kafka-console-consumer.sh --topic maxwell-events --from-beginning --bootstrap-server localhost:9092

ターミナルに次のような表示がされるはずです。

{"database":"maxwelltest","table":"Persons","type":"insert","ts":1602904030,"xid":17358,"commit":true,"data":{"PersonId":1,"LastName":"Erichsen","FirstName":"Tom","City":"Stavanger"}}

このようにMaxwell's Daemonは、DB挿入イベントをキャプチャして、イベントの詳細を含むJSON文字列をKafkaトピックに書き込みます。

MongoDBのセットアップ

MongoDBをローカルで実行するには、次のコマンドを実行してください。

docker run --name mongolocal -p 27017:27017 mongo:latest

Kafka Consumer

Kafka-consumerコードはgithubプロジェクトで公開されています。コードをダウンロードして、READMEの実行方法に関するセクションを参照してください。

最終テスト

セットアップが完成しました。MySQL DBにログインして、insert、delete、updateなどのコマンドを実行してください。セットアップが正しければ、mongodb auditlogデータベースにエントリができるはずです。それでは、楽しい監査を！

まとめ

この記事で紹介したシステムは、実際のデプロイメントで運用されていて、ユーザデータの付加的データソースを私たちに提供してくれています。ですが、当然のことながら、このアーキテクチャを採用する前には、注意しなければならないトレードオフがいくつか存在します。

1. インフラストラクチャのコスト — このセットアップを運用するためのインフラストラクチャが必要です。データはデータベースからKafkaへ、さらに別のDBへ、場合によってはバックアップへと、複数のホップを転送されるため、これがインフラストラクチャコストを押し上げることになります。
2. データが複数のホップを渡り歩くため、監査ログのリアルタイム性を維持することはできません。数秒から数分の遅れが発生します。"監査ログにリアルタイム性は必要ない"という主張も可能ですが、このデータをリアルタイム監視に使用する計画があるならば、この点は考慮する必要があります。
3. このアーキテクチャでは、データの変更はキャプチャできますが、誰がデータを変更したのかは分かりません。どのデータベースユーザがデータを変更したのか、ということも知りたいのであれば、この設計はこのままでは機能しないでしょう。

トレードオフをいくつか挙げましたが、記事の最後に、今回のセットアップのメリットをもう一度紹介したいと思います。おもなメリットは、

• 監査ログによるデータベースパフォーマンスへのオーバーヘッドを抑制しながら、マーケティングや警告を目的とした付加的データソースへのニーズを充足することができます。
• セットアップが容易で堅牢です — セットアップ時のコンポーネントの問題によってデータを損失することがありません。例えばMDに障害が発生しても、データがバイナリログファイルに残っているので、デーモンが再起動された時に、残された情報を読み込むことが可能になります。Kafkaブローカに障害が発生した場合には、MDがそれを検知して、binlogからの読み込みを停止します。Kafkaコンシューマがクラッシュしても、データはKafkaブローカに残っています。従って、最悪のケースにおいても、監査ログが遅延することはありますが、データを失うことはありません。
• セットアップが容易で、開発作業時間の多くを消費することがありません。

著者について

Vishal Sinha氏は、分散コンピューティングと大規模なスケーラブルシステムに深い経験と関心を持つ、熱意あるテクノロジストです。氏はインドを代表するユニコーン企業で、テクノロジ担当ディレクタとして働いています。ソフトウェア産業での16年を越えるキャリアの中で氏は、いくつかのMNCやスタートアップに参加し、さまざまな大規模システムを開発し、多くのソフトウェアエンジニアからなるチームを指導してきました。複雑な問題を解決することと、新たなテクノロジを試すことを楽しんでいます。