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コネクターを選択します。ドロップダウン横の作成ボタンをクリックするとポップアップで新規コネクターを素早く作成可能です。必要な設定パラメーターはコネクターの作成を参照してください。希望の書き込みモード（非同期または同期）を選択します。
有効なJSONオブジェクトにレンダリングされるメッセージテンプレートを設定します。
フォールバックアクション（任意）：メッセージ配信失敗時の信頼性向上のため、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（Time To Live）は、リクエストがバッファに入ってから有効とみなされる最大秒数を指定します。TTLを超えてバッファに残るか、ディスクログへの永続化応答やアックが期限内に得られない場合、リクエストは期限切れと判断されます。 | |
| ヘルスチェック間隔 | Sinkがディスクログの自動ヘルスチェックを行う間隔（秒）を指定します。 | 15 |
| 最大バッファキューサイズ | ディスクログSinkの各バッファワーカーがバッファリング可能な最大バイト数を指定します。バッファワーカーはデータを一時的に保持し、効率的なデータストリーム処理を実現します。システム性能やデータ送信要件に応じて調整してください。 | 256 |
| クエリモード | 同期（synchronous）または非同期（asynchronous）のリクエストモードを選択し、メッセージ送信を最適化します。非同期モードではディスクログへの書き込みがMQTTメッセージのパブリッシュ処理をブロックしませんが、クライアントがメッセージを受信してからディスクログへの書き込みが完了するまでの遅延が発生する可能性があります。 | Asynchronous |
| バッチサイズ | EMQXからディスクログへ一括書き込みするデータの最大バッチサイズを指定します。サイズを調整することでデータ転送の効率と性能を細かく制御できます。 バッチサイズを 1に設定すると、データレコードは個別に送信され、バッチ処理は行われません。一般的にバッチサイズは大きめに設定することが推奨されます。 | 1000 |
| インフライトウィンドウ | 「インフライトキューリクエスト」とは、送信済みで応答やアックをまだ受け取っていないリクエストを指します。この設定はSinkとディスクログ間の通信で同時に存在可能なインフライトリクエストの最大数を制御します。 リクエストモードが asynchronousの場合、このパラメーターは特に重要です。同一MQTTクライアントからのメッセージを厳密に順序処理する必要がある場合は、この値を1に設定してください。 | 100 |