Atlas Stream Processing のモニタリングとメトリクス

Atlas Stream Processing はモニタリングとアラートを提供し、ユーザーはパフォーマンスとステータスのインサイトを活用してワークフローを改善できます。

Atlas UI でストリームプロセッサのワークスペースを監視する

すべての Stream Processing ワークスペースについて、Atlas UI でストリームプロセッサを監視できます。

Atlas で、プロジェクトの Stream Processing ページに移動します。

まだ表示されていない場合は、プロジェクトを含む組織をナビゲーションバーの Organizations メニューで選択します。
まだ表示されていない場合は、ナビゲーションバーの Projects メニューからプロジェクトを選択します。
サイドバーで、 Streaming Data見出しの下のStream Processingをクリックします。

Atlas Stream Processingページが表示されます。

監視対象の Stream Processing ワークスペースを選択します。

監視対象の Stream Processing ワークスペースのペインで、Manageをクリックします。

[Monitoring] タブをクリックします。

[Monitoring] タブには、選択したストリームプロセッサに関するさまざまな実行時統計情報が表示されます（以下に限定されません）。

取り込まれたメッセージの数
正常に処理されたメッセージの数
デッド文字キューに送信されたメッセージの数

ソース接続が Apache Kafka の場合、トピックのパーティションにおける現在のオフセットとブローカー上の最新オフセットとの間のラグ、およびすべてのパーティションラグの合計を監視できます。

必要に応じて、モニタリングチャートをフィルタリングします。

チャートは、ストリームプロセッサ名、時間範囲、粒度でフィルタリングできます。

ストリームプロセッサの監視メソッド

Atlas Stream Processing は、ストリームプロセッサに関するオンデマンドレポート作成用に次のメソッドを提供します。

sp.processor.sample()メソッドを使用すると、現在実行中のストリームプロセッサによって出力されるドキュメントの小さなサンプルを確認できます。ユーザーは、サンプリングされた結果と期待される結果を比較して、集計パイプライン設計におけるエラーを診断できます。

メソッドは、選択したストリームsp.processor.stats() プロセッサに関する実行時間統計を返します。これを呼び出す方法については、「ストリームプロセッサの統計情報を表示する」を参照してください。

ストリームプロセッサメトリクス

次のメトリクスは、ストリームプロセッサのランタイム動作を説明します。 Atlas Stream Processingsp.processor.stats() Monitoringは、出力ドキュメント、Atlas UIのタブ、および Atlas Administration APIと Datadog および OpenTelemetry の統合を通じてそれらをレポートします。

出力ドキュメントには、次のフィールドがあります。

フィールド	タイプ	説明
`ns`	string	ストリームプロセッサが定義されている名前空間。
`stats`	オブジェクト	ストリームプロセッサの動作状態を説明するドキュメント。
`stats.name`	string	ストリームプロセッサの名前。
`stats.status`	string	ストリームプロセッサのエンジン状態。これは単一のワーカーでの内部実行状態を表します。このフィールドには、次の値を指定できます。 `starting` `running` `error` `stopping` これは、ストリームプロセッサーの Atlas コントロールプレーンのライフサイクル状態を報告する`sp.listStreamProcessors()`が返す `state` フィールドとは異なります。2 つの値は異なる可能性があります。例えば、プロセッサーのコントロールプレーン `state` が `STOPPING` であっても、エンジン `stats.status` はまだ `running` である場合や、プロセッサーがまだワーカーで実行されていないため、コントロールプレーン `state` が `PROVISIONING` であっても `stats.status` が値を返さない場合があります。
`stats.scaleFactor`	integer	サイズフィールドが表示される単位。 `1`に設定されている場合、サイズはバイト単位で表示されます。 `1024`に設定されている場合、サイズはキロバイト単位で表示されます。
`stats.inputMessageCount`	integer	ストリームに公開されたドキュメントの数。ドキュメントは、パイプライン全体を通過する場合ではなく、 `$source`ステージを通過するとストリームに「公開」されたと見なされます。
`stats.inputMessageSize`	integer	ストリームに公開されたバイト数またはキロバイト数。バイトは、パイプライン全体を通過する場合ではなく、 `$source`ステージを通過するとストリームに「公開」と見なされます。
`stats.outputMessageCount`	integer	ストリームによって処理されたドキュメントの数。ドキュメントはストリームによってパイプライン全体を通過すると、ストリームによって「処理済み」と見なされます。
`stats.outputMessageSize`	integer	ストリームによって処理されたバイト数またはキロバイト数。バイトはパイプライン全体を通過すると、ストリームによって「処理済み」と見なされます。
`stats.dlqMessageCount`	integer	デッド文字キューに送信されたドキュメントの数。
`stats.dlqMessageSize`	integer	デッド文字キューに送信されたバイト数またはキロバイト数。
`stats.changeStreamTimeDifferenceSecs`	integer	最新の変更ストリーム再開トークンによって表されるイベント時間とoplog 内の最新のイベントとの差（秒単位）。
`stats.changeStreamState`	token	最新の変更ストリーム再開トークン。変更ストリームソースを持つストリームプロセッサにのみ適用されます。
`stats.latency`	ドキュメント	ストリームプロセッサ全体のレイテンシ統計です。Atlas Stream Processing は、`verbose`オプションを指定した場合にのみ、このフィールドを返します。
`stats.latency.p50`	integer	過去 30 秒間に処理されたすべてのドキュメントの推定 50 パーセンタイルレイテンシ。パイプラインにウィンドウステージが含まれている場合、レイテンシ測定にはウィンドウの間隔が含まれます。例えば、`$tumblingWindow` ステージの間隔が 5 分の場合、レイテンシ測定にはその 5 分が含まれます。
`stats.latency.p99`	integer	過去 30 秒間に処理されたすべてのドキュメントの推定 99 パーセンタイルレイテンシ。パイプラインにウィンドウステージが含まれている場合、レイテンシ測定にはウィンドウの間隔が含まれます。例えば、`$tumblingWindow` ステージの間隔が 5 分の場合、レイテンシ測定にはその 5 分が含まれます。
`stats.latency.start`	datetime	直近の 30 秒測定ウィンドウが開始した時刻（ウォールタイム）。
`stats.latency.end`	datetime	直近の 30 秒間の測定ウィンドウが終了した時点のウォール時間。
`stats.latency.unit`	string	レイテンシが計測される時間の単位。この値は常に `microseconds` です。
`stats.latency.count`	integer	直近の 30 秒測定ウィンドウでストリームプロセッサが処理したドキュメント数。
`stats.latency.sum`	integer	直近の 30 秒測定ウィンドウで取得された、個々のレイテンシ測定値の合計（マイクロ秒単位）。
`stats.stateSize`	integer	Windowsがプロセッサの状態を保存するために使用するバイト数。
`stats.throttle.throttledTimeMs`	integer	ストリームプロセッサがアクティブなスロットリングに費やした累計時間（ミリ秒単位）。パイプラインに複数の `$throttle` ステージが含まれている場合、この値はステージごとの値の合計です。
`stats.throttle.throttleEvents`	integer	ストリームプロセッサが設定されたレート制限内に収まるようにメッセージを遅延させた回数。パイプラインに複数の `$throttle` ステージが含まれている場合、この値はステージごとの値の合計です。
`stats.watermark`	date	現在の浮動小数のタイムスタンプ。
`stats.addedParallelism`	integer	ストリームプロセッサの合計並列処理。 Atlas Stream Processing では、`1` より大きい `parallelism` 値を設定するすべてのステージにわたる `(parallelism - 1)` の合計としてこの値が計算されます。このフィールドはパイプライン全体を要約するため、`stats.operatorStats` には表示されません。 Atlas Stream Processing では、少なくとも 1 つのステージで `parallelism` の値が `1` より大きい場合にのみこのフィールドが返されます。例、`$source` `initialSync.parallelism2`ステージがの値を設定し、$lookup ステージがの`parallelism` 値を設定し、`4` $merge ステージが`parallelism` の`4` 値を設定する場合です。をクリックすると、`stats.addedParallelism` `(2 - 1) + (4 - 1) + (4 - 1)`は、または`7` を報告します。この値を使用して、構成した並列処理とストリームプロセッサ層で使用可能な並列処理を比較します。詳しくは、「 Atlas Stream Processing 階層選択ガイド」を参照してください。
`stats.operatorStats`	配列	プロセッサパイプライン内の各演算子の統計情報。 Atlas Stream Processing では、 `verbose`オプションで渡された場合にのみこのフィールドが返されます。 `stats.operatorStats` は、多数のコア`stats`フィールドの演算子ごとのバージョンを提供します。 `stats.operatorStats.name` `stats.operatorStats.inputMessageCount` `stats.operatorStats.inputMessageSize` `stats.operatorStats.outputMessageCount` `stats.operatorStats.outputMessageSize` `stats.operatorStats.dlqMessageCount` `stats.operatorStats.dlqMessageSize` `stats.operatorStats.latency` `stats.operatorStats.stateSize` `stats.operatorStats.throttle.throttledTimeMs` `stats.operatorStats.throttle.throttleEvents` `stats.operatorStats` には、次の一意のフィールドが含まれます。 `stats.operatorStats.maxMemoryUsage` `stats.operatorStats.executionTimeMillis` `stats.operatorStats` `verbose` オプションを渡し、プロセッサにウィンドウステージが含まれている場合、次のフィールドも含まれます。 `stats.minOpenWindowStartTime` `stats.maxOpenWindowStartTime` `stats.operatorStats` には、特定のソース演算子とシンク演算子に次のフィールドを含めることもできます。 `stats.operatorStats.targetStats`
`stats.operatorStats.maxMemoryUsage`	integer	演算子の最大メモリ使用量（バイトまたはキロバイト単位）。
`stats.operatorStats.executionTimeMillis`	integer	演算子の合計実行時間（ミリ秒）。
`stats.minOpenWindowStartTime`	date	最小オープンウィンドウの開始時刻。この値は任意です。
`stats.maxOpenWindowStartTime`	date	最大オープンウィンドウの開始時刻。この値は任意です。
`stats.operatorStats.targetStats`	配列	特定のソースとシンク演算子のターゲットごとの統計。この配列の各要素は、入力コレクションや出力コレクションやApache Kafkaトピックなど、単一の入力ターゲットまたは出力ターゲットを表すドキュメントです。演算子に応じて、各ドキュメントには次のフィールドのサブセットが含まれます。 Apache Kafka `$source`などのソース演算子または変更ストリームソースの場合: MongoDB ターゲット`db` と `coll` か、Apache Kafka ターゲット`topic` のいずれかになります。 `inputMessageCount` `inputMessageSize` MongoDB `$merge` または `$emit` シンクのようなシンク演算子の場合: MongoDB ターゲット`db` と `coll` か、Apache Kafka ターゲット`topic` のいずれかになります。 `outputMessageCount` `outputMessageSize` Atlas Stream Processing は、Apache Kafka `$source`、MongoDB`$merge` などの一部のストリーム集計でターゲットごとの統計を収集します。最大で100件の明確に区別されたターゲットのターゲットごとの統計を記録します。その後、ストリームプロセッサは `targetStats` への新規エントリの追加を停止しますが、演算子ごとの統計の集計は継続して更新します。
`stats.kafkaPartitions`	配列	Apache Kafkaエージェントのパーティションのオフセット情報。`kafkaPartitions` はApache Kafkaソースを使用する接続にのみ適用されます。
`stats.kafkaPartitions.partition`	integer	Apache Kafkaトピックのパーティション番号。
`stats.kafkaPartitions.currentOffset`	integer	指定されたパーティションに対するストリームプロセッサのオフセット。この値は、ストリームプロセッサが処理した以前のオフセットに`1`を加えた値に等しくなります。
`stats.kafkaPartitions.checkpointOffset`	integer	ストリームプロセッサがApache Kafkaプロバイダーと指定されたパーティションのチェックポイントに最後にコミットしたオフセット。このオフセットを介するすべてのメッセージは、最後のチェックポイントに記録されます。
`stats.kafkaPartitions.isIdle`	ブール値	パーティションがアイドル状態であるかどうかを示すフラグ。この値はデフォルトで `false` です。

Atlas Stream Processing ログ

Atlas Stream Processing では、次の 2 種類のストリーム処理ワークスペースのアクティビティログが提供されます。

操作ログ（主に以下の動作を追跡するログ）
個々のストリームプロセッサ。
監査ログ（主に認証とセキュリティを追跡）
ストリーム処理ワークスペースレベルでのアクティビティ。

垂直オートスケーリングが有効になっているプロセッサーの場合、操作ログには、各スケールアップイベントとスケールダウンイベントがレコードされます。それには、プロセッサーが移動した元の階層と移動先の階層、およびイベントをトリガーしたメトリクスが含まれます。

Atlas Stream Processing の運用ログまたは監査するログをダウンロードするには:

Atlas で、プロジェクトの Stream Processing ページに移動します。

まだ表示されていない場合は、プロジェクトを含む組織をナビゲーションバーの Organizations メニューで選択します。
まだ表示されていない場合は、ナビゲーションバーの Projects メニューからプロジェクトを選択します。
サイドバーで、 Streaming Data見出しの下のStream Processingをクリックします。

Atlas Stream Processingページが表示されます。

ログをダウンロードするストリーム処理ワークスペースのペインに移動し、省略記号をクリックします。

ドロップダウンメニューで [Download Logs] を選択します。

モーダルウィンドウで、ダウンロードするログのタイプを選択します。

Stream processorフィールドに、ログをダウンロードするストリームプロセッサの名前を指定します。すべてのストリームプロセッサのログをダウンロードするには、このフィールドを空白のままにします。

[Time Period ドロップダウンから、ログをダウンロードする間隔を選択します。

[Download Logs] ボタンをクリックします。

注意

ログがクエリ可能になるまでには、作成後から最大 5 分かかります。

デッド文字キュー

デッドレターキュー (DLQ) 内のドキュメントを検討して、ストリームプロセッサーの障害モードを分析することができます。DLQ に書き込まれたドキュメントは、以下の形式をとります。

{
  "processorName": "<stream-processor-name>",
  "instanceName": "<workspace-name>",
  "dlqTime": {
    "$date": "<datetime>"
  },
  "operatorName": "<stream-operator-name>",
  "errInfo": {
    "reason": "<error-description>"
  },
  "doc": {
    "_id": {
      "$oid": "<object-id>"
    },
    . . .
  },
  "_stream_meta": {
    "<stage-metadata>": {
      "<metadata-field>": "<metadata-value>",
      . . .
    },
    . . .
  }
}

DLQ ドキュメントには、次のフィールドが含まれています。

フィールド	タイプ	条件付き	説明
processorName	string	常に	ドキュメントの発生元となるストリームプロセッサの名前。
instanceName	string	条件付き	プロセッサが実行される Stream Processing ワークスペースの名前。プロジェクトに複数のワークスペースが含まれている場合にのみ書き込まれます。
dlqTime.$date	ISODate	常に	Atlas Stream Processingがデッドレターキュー (DLQ)にドキュメントを登録した日時。
operatorName	string	常に	Atlas Stream Processing のエラーが発生した集計演算子またはステージ。
errInfo.reason	string	常に	ドキュメントプロセシングの失敗の理由。
doc	ドキュメント	常に	この障害イベントによりDLQ に書込まれた完全なドキュメント。
_stream_meta	ドキュメント	常に	ストリームメタデータを含むドキュメント。このドキュメントの内容は、指定された入力に対する $meta ステージの出力と同じです。詳細については、$meta 構文リファレンスを参照してください。

サードパーティメトリクス統合

Atlas Stream Processing はサードパーティーのメトリクス統合をサポートしており、ユーザーはカスタムロジックやビューを開発することなく、ストリームプロセッサーのパフォーマンスをレコードして分析できます。

DataDog は、接頭語 mongodb.atlas.stream_processing を使用して Atlas Stream Processing メトリクスを追跡します。統合を設定し、利用可能なメトリクスを表示するには、「DataDog との統合」を参照してください。

OpenTelemetry (OTel) コレクターにメトリクスを送信するには、「OpenTelemetry (OTel) との統合」を参照してください。

ストリームプロセッサアラート

Atlas Stream Processing は、プロセッサの状態が変化したとき、またはプロセッサがさまざまな取り込みや出力のしきい値に達したときにアラートをトリガーします。使用可能な Atlas Stream Processing アラートのリストについては、「 Atlas Stream Processing アラート」を参照してください。アラート構成の詳細については、「アラート設定の構成」を参照してください。

次の方法で、Atlas Stream Processing アラートを対象にすることができます。

プロジェクト内のすべてのストリームプロセッサ
設定された述語に一致する Stream Processing ワークスペース内のすべてのストリームプロセッサ
設定された述語に一致する名前を持つすべてのストリームプロセッサ

すべてのストリームプロセッサ以外のターゲットに対して、同じアラートに複数のターゲットを設定できます。

Atlas Stream Processing は、デフォルトで Stream Processor State is failed アラートを構成します。これはプロジェクトレベルのアラートであるため、構成されているプロジェクト内の任意の stream processing ワークスペースで実行中のストリームプロセッサに適用されます。

戻る

Stream Processing 権限アクション

制限

Atlas Stream Processing のモニタリングとメトリクス

Atlas UI でストリーム プロセッサのワークスペースを監視する

Atlas で、プロジェクトの Stream Processing ページに移動します。

監視対象の Stream Processing ワークスペースを選択します。

[Monitoring] タブをクリックします。

必要に応じて、モニタリング チャートをフィルタリングします。

ストリーム プロセッサの監視メソッド

ストリーム プロセッサ メトリクス

Atlas Stream Processing ログ

Atlas で、プロジェクトの Stream Processing ページに移動します。

ドロップダウン メニューで [Download Logs] を選択します。

[Download Logs] ボタンをクリックします。

デッド文字キュー

サードパーティメトリクス統合

ストリーム プロセッサ アラート

Atlas UI でストリームプロセッサのワークスペースを監視する

必要に応じて、モニタリングチャートをフィルタリングします。

ストリームプロセッサの監視メソッド

ストリームプロセッサメトリクス

ドロップダウンメニューで [Download Logs] を選択します。

ストリームプロセッサアラート