MQTTデータをディスクログに取り込む
ディスクログのデータ統合により、EMQXはイベントデータをJSON Lines形式でディスクに永続化できます。これは従来のローテーティングログファイルに似ており、トラブルシューティングや履歴追跡のための長期的なイベント保持を可能にします。
本ページでは、EMQXとディスクログ間のデータ統合について詳細に解説し、ルールおよびSinkの作成方法について実践的なガイダンスを提供します。
動作概要
EMQXはエラーや警告、システムアクティビティなどの運用イベントを監視するための組み込みシステムログを備えていますが、ディスクログ統合は異なる目的を持ちます。実際のMQTTメッセージデータやクライアントレベルのイベントをディスクに永続化し、保持およびオフライン処理を可能にします。
ディスクログ統合はEMQXのルールエンジンとSink機構を用いて実装されており、ユーザーはどのデータをキャプチャし、どのように保存するかを正確に定義できます。
- ルールはMQTTメッセージやクライアントイベントから関心のあるデータをフィルタリング、変換、抽出します。
- ルールにディスクログSinkを紐付け、データの保存方法と保存先を定義します。Sinkはフォーマット済みのデータ（JSON形式）を対応するコネクターに転送します。
- ディスクログコネクターはファイルシステムへの物理的な書き込みを管理します。ログファイルパスの設定やログファイルのローテーションポリシーなどを取り扱います。
- ルールがトリガーされデータがSinkに渡されると、Sinkは設定されたコネクターを呼び出し、JSON Lines形式で設定されたベースファイルパスに基づくローテーティングファイルに書き込みます。これにより標準ツールや下流のデータシステムでの利用が容易になります。
ログローテーション
ディスクログ統合はローカルファイルシステム上の一連のローテーティングファイルにメッセージを書き込みます。設定されたログファイルパスはディレクトリではなくベースファイルパスです。例えば、ベースパスが/var/log/emqx/mqtt-trace.logの場合、実際のログファイルはmqtt-trace.log.1のように数値のサフィックスが付きます。
ディスクログはサイズベースと時間ベースのローテーションをサポートし、サイズベースは常に有効です。
- 設定された最大ファイルサイズに達するとEMQXは新しいファイルを開いて書き込みを継続します。
- 最大ファイル数に達すると、最も古いファイルの内容を破棄し、新しいエントリ用にそのファイルを再利用します。
- 各ログファイルには、たとえ単一のエントリがファイルサイズ制限を超えていても、少なくとも1つの完全なエントリが含まれることが保証されます。
時間ベースのローテーションは各時間または日付の境界で別のファイルセットを開始します。以下のコネクター設定項目を指定してください。
| 項目 | 設定キー | 説明 | デフォルト |
|---|---|---|---|
| ローテーション周期 | rotation.period | hourまたはdayを選択すると各周期境界でファイルセットを開始します。None（none）を選択するとサイズベースのみのローテーションになります。 | None（none） |
| 保持期間 | rotation.retention_period | EMQXが過去の周期のファイルセットを保持する期間を指定します。コネクター起動時および各周期ローテーション後に期限切れのファイルセットを削除します。rotation.periodがnoneの場合は無効です。 | infinity |
| ローテーションタイムゾーン | rotation.timezone | 周期境界とファイル名のタイムスタンプを決定するためのタイムゾーンを指定します。UTC、local、または+02:00のような固定UTCオフセットが指定可能です。 | UTC |
時間ベースのローテーションが有効な場合、EMQXはファイル拡張子の前にハイフン（-）とYYYYMMDDHH形式のタイムスタンプを挿入します。時間単位のファイルは周期の時間を使用し、例：mqtt-trace-2026062413.log.1。日単位のファイルは時間を00に固定し、例：mqtt-trace-2026062400.log.1となります。.Nサフィックスは常に付与されます。各周期のファイルセットには.idxおよび.sizの管理ファイルも含まれます。
最大ファイルサイズおよび最大ファイル数の設定は各周期ごとに適用されます。周期内でファイルの上書きを防ぐため、1時間または1日の最大想定データ量に十分な容量を設定してください。各周期の概算容量は最大ファイルサイズ×最大ファイル数です。
EMQXはコネクターのヘルスチェック時に周期境界を検知します。そのため、新しい周期のファイルは境界から設定されたresource_opts.health_check_interval秒以内に出現します。
以下はbase.hoconで日単位ローテーションかつ30日間保持を定義した例です。
connectors.disk_log.my_disk_log {
filepath = "/var/log/emqx/mqtt-trace.log"
max_file_size = "1GB"
max_file_number = 24
rotation {
period = day
retention_period = "30d"
timezone = "UTC"
}
}重要なお知らせ
稼働中のディスクログファイルに対して外部のlogrotateのcopytruncateオプションを使用しないでください。外部での切り詰めは.sizファイルで管理されているサイズ情報をリセットしないため、予期しないローテーションが発生する恐れがあります。時間ベースのローテーションは内蔵機能を利用してください。
特長と利点
ディスクログ統合はMQTTメッセージの永続化に対して柔軟で軽量、かつローカルファーストなソリューションを提供します。主な特長と利点は以下の通りです。
- 細かなデータ制御：SQLベースのルールで必要なメッセージやイベントのみをログに記録。ログ前に変換、フィルタリング、付加処理も可能です。
- 構造化された出力形式：JSON Lines形式でデータを保存し、機械処理を容易にします。
- 軽量かつ自己完結型：外部ストレージやデータベースとの接続は不要です。
- 可観測性とデバッグ：メッセージレベルの可視化を実現し、トラブルシューティングや監査に役立ちます。EMQXのシステムログと補完的に機能し、システムイベントではなくデータフローを記録します。
はじめる前に
本節ではEMQXでディスクログSinkを作成する前に必要な準備について説明します。
前提条件
ログディレクトリの作成
EMQXホスト上にログファイル保存用の書き込み可能なディレクトリを作成してください。EMQXのシステムユーザーがこのディレクトリに対して読み書き権限を持っている必要があります。
コネクターの作成
ディスクログSinkを追加する前に、対応するコネクターを作成します。
- EMQXダッシュボードの左ナビゲーションメニューでIntegration -> Connectorsをクリックします。
- 右上のCreateボタンをクリックします。
- コネクタータイプとしてDisk Logを選択し、Nextをクリックします。
- コネクター名を半角英数字の組み合わせで入力します。ここでは
my-disk-logと入力します。 - コネクターのパラメーターを入力します。
- Log Filepath：ログファイルのベースパス。例：
/var/log/emqx/mqtt-trace.log。ファイルを含むディレクトリはEMQXシステムユーザーが書き込み可能である必要があります。 - Maximum File Size：ローテーション前の各ファイルの最大サイズ。注意：各ログには少なくとも1つのエントリが書き込まれるため、単一エントリがこの値を超える場合は最終ファイルサイズも超過します。
- Maximum Number of Files：各ファイルセットの最大ファイル数。時間ベースローテーションが有効な場合、この制限は各時間または日ごとに適用されます。
- Rotation Period：
hourまたはdayを選択すると各周期境界でファイルセットを開始します。Noneを選択するとサイズベースのみのローテーションになります。 - Retention Period：過去の周期のファイルセットを保持する期間を指定します。デフォルトは
infinityです。 - Rotation Timezone：周期境界とファイル名タイムスタンプの判定に使用するタイムゾーンを
UTC、local、または+02:00のような固定UTCオフセットで指定します。詳細はログローテーションを参照してください。
- Log Filepath：ログファイルのベースパス。例：
- Createをクリックする前に、Test Connectivityをクリックしてコネクターが設定されたパスにログを書き込めるかテストできます。
- 画面下部のCreateボタンをクリックしてコネクター作成を完了します。
これでコネクターの作成が完了しました。次にルールとSinkを作成し、ディスクログに書き込むデータを指定します。
ディスクログSink付きルールの作成
本節では、EMQXでソースMQTTトピックt/#からメッセージを処理し、処理結果を設定済みのSink経由でローカルログファイルに書き込むルールの作成方法を示します。
ダッシュボードのIntegration -> Rulesページに移動します。
右上のCreateボタンをクリックします。
ルールIDに
my_ruleを入力し、SQLエディターに以下のルールSQLを入力します。sqlSELECT * FROM "t/#"TIP
SQLに不慣れな場合は、SQL ExamplesやEnable DebugをクリックしてルールSQLの学習やテストが可能です。
アクションを追加し、Action Typeドロップダウンリストから
Disk Logを選択します。アクションのドロップダウンはデフォルトのcreate actionのままにするか、既存のDisk Logアクションを選択します。ここでは新しいSinkを作成してルールに追加します。Sinkの名前と説明を入力します。
先に作成した
my-disk-logコネクターをコネクタードロップダウンから選択します。ドロップダウン横の作成ボタンをクリックするとポップアップで新規コネクターを素早く作成可能です。必要な設定パラメーターはコネクターの作成を参照してください。希望のWrite Mode（非同期または同期）を選択します。
有効なJSONオブジェクトにレンダリングされるMessage Templateを設定します。
フォールバックアクション（任意）：メッセージ配信失敗時の信頼性向上のため、1つ以上のフォールバックアクションを定義できます。プライマリSinkがメッセージ処理に失敗した場合にこれらのアクションがトリガーされます。詳細はフォールバックアクションを参照してください。
詳細設定を展開し、必要に応じて詳細設定オプションを構成します（任意）。詳細は詳細設定を参照してください。
残りの設定はデフォルト値のままにし、CreateボタンをクリックしてSink作成を完了します。作成成功後、ルール作成画面に戻り、新しいSinkがルールアクションに追加されます。
ルール作成画面でCreateボタンをクリックし、ルール作成全体を完了します。
これでルールが正常に作成されました。Rulesページで新規ルールを確認でき、**Actions (Sink)**タブで新しいディスクログSinkを確認できます。
また、Integration -> Flow Designerをクリックするとトポロジーが表示されます。トポロジーはトピックt/#のメッセージがルールmy_ruleで解析され、ディスクログに書き込まれる流れを視覚的に示します。
ルールのテスト
本節では、直接アップロード方式で設定したルールのテスト方法を示します。
MQTTXを使ってトピックt/1にメッセージをパブリッシュします。
mqttx pub -i emqx_c -t t/1 -m '{ "msg": "Hello Disk Log" }'数件メッセージを送信した後、設定したディスクログディレクトリ内の最終更新ファイルを確認し、生成されたイベントの内容を確認してください。
詳細設定
本節ではディスクログSinkの詳細設定オプションについて説明します。ダッシュボードのSink設定画面で詳細設定を展開し、用途に応じて以下のパラメーターを調整できます。
| 項目名 | 説明 | デフォルト値 |
|---|---|---|
| バッファプールサイズ | EMQXとディスクログ間のデータフローを管理するバッファワーカープロセスの数を指定します。これらのワーカーはデータを一時的に保持・処理し、ディスクログへの送信を最適化しスループット向上に寄与します。 | 16 |
| リクエストTTL | バッファに入ったリクエストが有効とみなされる最大時間（秒）を指定します。リクエストがこのTTLを超えてバッファに滞留するか、ディスクログへの永続化応答やアックを受け取れない場合、リクエストは期限切れと見なされます。 | |
| ヘルスチェック間隔 | Sinkがディスクログの自動ヘルスチェックを行う間隔（秒）を指定します。 | 15 |
| 最大バッファキューサイズ | 各バッファワーカーがディスクログSink内でバッファリング可能な最大バイト数を指定します。バッファワーカーはデータを一時保持し、効率的にデータストリームを処理します。システム性能やデータ転送要件に応じて調整してください。 | 256 |
| クエリモード | synchronous（同期）またはasynchronous（非同期）のリクエストモードを選択し、メッセージ送信を最適化します。非同期モードではディスクログへの書き込みがMQTTメッセージのパブリッシュ処理をブロックしませんが、クライアントがメッセージ受信後にディスクログへ書き込まれる前の状態になる可能性があります。 | Asynchronous |
| バッチサイズ | EMQXからディスクログへ一度に書き込むデータの最大バッチサイズを指定します。サイズを調整することでデータ転送の効率や性能を微調整できます。 バッチサイズを 1に設定すると、データレコードは個別に送信されバッチ処理されません。このアクションは大きなバッチサイズでの運用に適しています。 | 1000 |
| インフライトウィンドウ | 「インフライトキューリクエスト」とは開始されたが応答やアックをまだ受け取っていないリクエストを指します。この設定はSinkとディスクログ間の通信で同時に存在可能なインフライトリクエストの最大数を制御します。 リクエストモードが asynchronousの場合、このパラメーターは特に重要です。同一MQTTクライアントからのメッセージを厳密に順序処理する必要がある場合は、値を1に設定してください。 | 100 |