Azure OpenAI Service Dev Day に登壇しました

<html><head></head><body><h1 id="はじめに">はじめに</h1>
こんにちは、プレイドのCore Platformチームに所属している西村です。今回のブログではGCE上でセルフホストしていたMongoDBインスタンスをMongoDB Atlasに移行した際の取り組みについて紹介します。移行に伴う要件や制約を考慮した結果、今回はあえてMongoDB固有の機能を使わずにGKEのJobを使って愚直にデータ移行する方法を採用したので、同じような移行作業を行う場合は参考になれば幸いです。
<h2 id="背景">背景</h2>
プレイドではDBとして多くの部分で<a href="https://www.mongodb.com/ja-jp/atlas" target="_blank" rel="noreferrer">MongoDB Atlas</a>を使用しています。MongoDB Atlasでは定期的にサポートバージョンが更新され、プレイドでもそれに合わせて適宜バージョンアップ対応を行っています。<a href="https://tech.plaid.co.jp/internship-mongodb-update">前回のバージョンアップ作業</a>は2023年末から2024年3月頃にかけて実施したのですが、そこで長らくGCE上で稼働していた古いバージョンのMongoインスタンスをこの機会にMongoDB Atlasに移行することにしました。
このインスタンスはOplogというKARTEの管理画面の操作履歴を保存しているもの（以降、旧Oplog）で、通常はほとんど参照されないデータなのでこれまでのバージョンアップ作業でも見落とされてしまっていました。しかし今回のバージョンアップでこのインスタンスで稼働しているMongoのバージョン（3.x）と更新後のバージョン（5.x）の互換性がなくなるため、こちらもバージョンアップ対応するかMongoDB Atlasに移行するか選択する必要がありました。今後のバージョンアップ対応や管理の煩雑さを考えて、この機に旧Oplogを廃止してMongoDB Atlas（以降、新Oplog）に移行する判断をしました。
<h2 id="移行方法の検討">移行方法の検討</h2>
移行にあたってまず要件を決める必要があります。今回の移行プロジェクトの要件は以下のように定めました。
<ul>
<li>旧Oplogの過去データをすべて新Oplogに移行する</li>
<li>サービスは無停止でデータおよび書き込み先の移行を行う</li>
<li>想定外のトラブルがあってもデータの欠損が極力発生しないようにする</li>
<li>移行完了後に旧Oplogは廃止する</li>
</ul>
上記の要件を満たすために以下の作業を順次実施することにしました。
<ol>
<li>MongoDB Atlasに移行用Clusterを作成する（新Oplog）</li>
<li>旧Oplogの書き込みと同時に新Oplogにもデータを書き込む2重書き込みの仕組みを実装する</li>
<li>2重書き込みが問題なくできていることを確認し、旧Oplogのデータを新Oplogに移行する</li>
<li>データ移行完了後、旧Oplogへの書き込みを停止しインスタンスを破棄する</li>
</ol>
図にすると以下のようなイメージです。 
<img src="https://ik.imagekit.io/plaid/ee37750e-63d1-47ac-9991-552cf0506361/migrate-oplog-2.png" alt="migrate-oplog-2.png">
以下ではそれぞれの手順で実施した内容について要点を解説します。
<h3 id="旧Oplogの書き込みと同時に新Oplogにもデータを書き込む2重書き込みの仕組みを実装する">旧Oplogの書き込みと同時に新Oplogにもデータを書き込む2重書き込みの仕組みを実装する</h3>
DBの移行にあたって必要な作業は主にデータ移行と接続先の変更の2つです。論理的には旧Oplogに書き込んでいる箇所の書き込み先を新しく作成した新Oplogに変更すれば新規に書き込むデータに関しては移行できるはずですが、いきなり接続先を切り替えると不具合があって書き込みができないなどのトラブルが発生した場合にデータが欠損してしまうため、慎重を期すために従来通り旧Oplogへの書き込みをしつつ新たに新Oplogへも同時に書き込む仕組みを実装し、既存の仕組みに影響を与えないようにしました。
<h3 id="2重書き込みが問題なくできていることを確認し、旧Oplogのデータを新Oplogに移行する">2重書き込みが問題なくできていることを確認し、旧Oplogのデータを新Oplogに移行する</h3>
2重書き込みについては特に大きな問題もなく実装することができました。これは一部のシステムを除いて旧Oplogを書き込む処理は社内の共通ライブラリ経由で実行されており、書き込み先の設定を変えて新Oplogへの書き込み処理を追加するだけで済んだためです。
あとは旧Oplogのデータを移行すれば必要な作業はほぼ終わりだったのですが、データ移行処理がなかなかうまくいかず試行錯誤が必要でした。その試行錯誤について詳しく見る前に、まずはMongoDBで推奨されているデータ移行方法と、なぜ今回それらの方法を採用しなかったのか簡単に説明します。
<h3 id="MongoDB標準のデータ移行方法">MongoDB標準のデータ移行方法</h3>
MongoDB標準のデータ移行方法としてJSON/CSVでの<a href="https://www.mongodb.com/docs/compass/current/import-export/" target="_blank" rel="noreferrer">Import/Export</a>と<a href="https://www.mongodb.com/ja-jp/products/platform/migrate" target="_blank" rel="noreferrer">Live Migration</a>があります。それぞれについて簡単に説明します。
<h4 id="Import/Export">Import/Export</h4>
Import/Exportは<a href="https://www.mongodb.com/ja-jp/products/tools/compass" target="_blank" rel="noreferrer">MongoDB Compass</a>での操作もしくはCLIでそれぞれに対応したコマンドを実行することでCollectionのデータを指定したフォーマットでExportおよびImportすることができる機能です。前回のMongoバージョンアップ作業で4.x→5.xでも想定外の微妙な挙動の差異があったが今回は3.x→5.xなので動作差分が読めないこと、本番データ量が膨大（約2.3億レコード、500GB以上）で一括Importに懸念があることなどから今回は採用しないことにしました（とはいえ一度試してみてもよかったかも）。
<h4 id="Live Migration">Live Migration</h4>
Live MigrationはMongoDB Atlasで提供されている機能で、コンソールから設定するだけでオンプレミスも含めて任意の環境からMongoDB AtlasのClusterにデータ移行できるという機能です。Live Migrationではデータ移行中も移行元DBへの読み取りおよび書き込みが可能ですが、別のチームで試してみたところ断続的に常に書き込みが発生しているシステムだとデータ移行が完了できないという特性があり、完全に無停止でデータを移行しきるのは難しいとのことでした。
上記のようにMongoDBの標準的なデータ移行方法では今回の要件を満たすことが難しそうだったため、旧OplogからCursorでデータを取得し新Oplogに書き戻すという愚直な方針を採用することにしました。
<h3 id="GKEでデータ移行用のJobを実行する">GKEでデータ移行用のJobを実行する</h3>
プレイドではアプリケーションの実行環境として多くの場所でGKEを使用しています。既存のデプロイの仕組みを流用するため、今回はデータ移行用のスクリプトをGKEのJobで実行することにしました。データ移行にあたっては以下の手順で段階的に進めていきました。
<ol>
<li>Dev環境にテスト用Clusterを作成し、少量のデータを移行する</li>
<li>Evaluation環境に移行用Clusterを作成し、Evaluationの旧Oplogのデータを移行する</li>
<li>Production環境に移行用Clusterを作成し、Productionの旧Oplogのデータを移行する</li>
</ol>
以下で各手順で試した内容と結果を説明します。
<h4 id="Dev環境にテスト用Clusterを作成し、少量のデータを移行する">Dev環境にテスト用Clusterを作成し、少量のデータを移行する</h4>
まずは旧Oplogから取得した任意のデータを新Oplogに書き込めることを確認するため、Dev環境で作成したテスト用ClusterにEvaluation環境の旧Oplogの1日分のデータを書き込んでみることにしました。データ量としては約5000〜6000レコード程度で、以下のような移行スクリプトで問題なく書き込めることが確認できました（サンプル用に簡略化しています）。
<pre><code class="language-typescript hljs">(async () =&gt; {
 console.log('Start: oplog-migration');
 const oplogGce = OperationLog(...);
 const oplogAtlas = OperationLogAtlas(...);

 // 移行時点以前のデータを抽出する（日付は仮）
 const dateBegin = new Date('2024-04-01T00:00:00.000+00:00');
 const dateEnd = new Date('2024-04-02T00:00:00.000+00:00');
 const cursor = oplogGce.find({ timestamp: { $gte: dateBegin, $lt: dateEnd } }).cursor({ batchSize: 5000 });

 cursor.on('data', (doc: any) =&gt; {
 const oplog = new oplogAtlas(doc.toObject());
 oplog.save();
 });

 cursor.on('end', () =&gt; {
 console.log('End: oplog-migration');
 });
})();
</code></pre>
<h4 id="Evaluation環境に移行用Clusterを作成し、Evaluationの旧Oplogのデータを移行する">Evaluation環境に移行用Clusterを作成し、Evaluationの旧Oplogのデータを移行する</h4>
テスト用Clusterへの書き込みが問題なく行えたので次はEvaluationの過去データ（約300万件、5GB程度）を移行するために日付の指定と書き込み先を変えて先ほどのJobを動かしたところOOM(Out Of Memory)エラーが発生するようになりました。
お気づきの方も多いかと思いますが、先ほどのコードではCurosrで取得したデータを1件ずつ書き込んでいたためデータ件数が多くなると書き込み処理が追いつかずOOMが発生してしまいます。今回のケースだと数万件程度処理が進んだ段階でOOMが発生しました。OOMを避けるためバッチサイズごとにバルクで書き込むように修正することでOOMの発生を抑えることができました。
<pre><code class="language-typescript hljs">(async () =&gt; {
 console.log('Start: oplog-migration');
 const oplogGce = OperationLog(...);
 const oplogAtlas = OperationLogAtlas(...);

 // 移行対象のデータを抽出する（日付は仮）
 const date = new Date('2024-04-01T00:00:00.000+00:00');
 const batchSize = 5000;
 const cursor = oplogGce.find({ timestamp: { $lte: date } }).cursor({ batchSize });

 let bulkOps: any[] = [];
 cursor.on('data', async (doc: any) =&gt; {
 const oplog = new oplogAtlas(doc.toObject());
 bulkOps.push({ insertOne: { document: oplog } });

 if (bulkOps.length === batchSize) {
 await oplogAtlas.bulkWrite(bulkOps);
 bulkOps = [];
 }
 });

 cursor.on('end', async () =&gt; {
 if (bulkOps.length &gt; 0) {
 await oplogAtlas.bulkWrite(bulkOps);
 }
 console.log('End: oplog-migration');
 });
})();
</code></pre>
<h5 id="途中でJobが止まる問題が発生">途中でJobが止まる問題が発生</h5>
前項のようにCurosrで取得したデータをバルクでまとめて処理することで安定的にJobを実行することができるようになりましたが、今度は20万件程度のデータを処理した時点で、残りの処理が完了する前にJob自体が終了してしまうという問題に遭遇しました。ここに至るまでにもOOM対策としてヒープメモリを追加したりCPUスペックの強化、 <code>async/await</code> で新Oplogへの書き込み結果を待つなどの試行錯誤をしてきたのですが、エラーログやGKEのメトリクスを見ても特に問題はなく、Jobとしてはすべての処理が完了して正常終了しているように見えました。
原因を調べるために調査用のログを仕込んで観察してみると、どうやら新Oplogへ書き込んでいる箇所のawaitが効いていなさそうだということがわかりました。つまり新Oplogへの書き込み箇所でawaitすることで書き込み処理の完了を待っているつもりだったのですが、Cursorの <code>on('data', ...)</code> はデータがある限り実行され続けてしまうため、新Oplogへの書き込みが完了する前にCursorの読み込みが終わってしまうという状態になっていそうだということがわかりました。
上記の現象の対策として新Oplogへの書き込み中はCursorの読み込みを一時停止し完了後に再開することで、無事に意図通りの挙動にすることができました。最終的なコードは以下のようになりました。
<pre><code class="language-typescript hljs"> let bulkOps: any[] = [];
 cursor.on('data', async (doc: any) =&gt; {
 const oplog = new oplogAtlas(doc.toObject());
 bulkOps.push({ insertOne: { document: oplog } });

 if (bulkOps.length === batchSize) {
 cursor.pause(); // 書き込み前にCursorの読み込みを中断
 await oplogAtlas.bulkWrite(bulkOps);
 bulkOps = [];
 cursor.resume(); // 書き込み完了後にCursorの読み込みを再開
 }
 });
</code></pre>
<h4 id="Production環境に移行用Clusterを作成し、Productionの旧Oplogのデータを移行する">Production環境に移行用Clusterを作成し、Productionの旧Oplogのデータを移行する</h4>
前項まで見てきたようにいろいろな落とし穴にハマりつつ、なんとか安定的にJobを動かしCursorで逐次的にデータを取得しMongoDB AtlasのClusterに大量データを書き込むという一連の処理を実装することができました。最後の仕上げとしてProduction環境の旧Oplogから新Oplogへのデータ移行Jobを実行して無事に全データを移行することができました。ちなみにProductionのデータは5年以上分、500GB程度ありJobの完了までに27時間程度かかりました。
<h2 id="まとめ">まとめ</h2>
本ブログではGCE上で稼働しているMongoDBインスタンスを無停止でMongoDB Atlasに移行した取り組みを紹介しました。今回の取り組みではサービス固有の機能に依存しない汎用的な手法で移行を行ったので、似たような作業をされる場合の一つの方法として参考になれば幸いです。
</body></html>

GCE上でセルフホストしていたMongoDBインスタンスをサービス無停止でMongoDB Atlasに移行した取り組みを紹介します

セルフホストしているMongoDBをMongoDB Atlasに移行した話