Sumo Logicでデータ取り込みが継続していることを確認する

本日はタイトルの通り、Sumo Logic のデータ取り込みが正しく行われていることを監視するための仕組みを実装するにはどうしたらいいかご紹介したいと思います。

本記事は、基本的に Sumo Logic のこちらの公式ドキュメントの内容をもとに書かれています。

いきなり答え

Sumo Logic のクエリ画面を開いて、

下記のようなクエリを作ります。

_index=sumologic_volume sizeInBytes _sourceCategory="collector_volume"
| parse regex "\"(?<collector>[^\"]+)\"\:\{\"sizeInBytes\"\:(?<bytes>\d+),\"count\"\:(?<count>\d+)\}" multi
| first(_messagetime) as MostRecent, sum(bytes) as TotalVolumeBytes by collector
| formatDate(fromMillis(MostRecent),"yyyy/MM/dd HH:mm:ss") as MostRecentTime
| toMillis(now()) as currentTime
| formatDate(fromMillis(currentTime),"yyyy/MM/dd HH:mm:ss") as SearchTime
| (currentTime-MostRecent) / 1000 / 60 as mins_since_last_logs
| where mins_since_last_logs >= 60
| fields -mostrecent, currenttime
| format ("%s Has not collected data in the past 60 minutes", collector) as message
| where collector = <collector名>

一番最後の行の<collector名>にデータ取り込みが止まっていないかを監視したいコレクター名をします。

今回は「test_」というコレクターにデータの欠損が発生していないかを調べることにしています。検索する範囲は24時間ぐらいが適切でしょう。
クエリを実行した後、「Aggregates」のところに何も出力がされていない場合は、欠損が無いということになります。

このクエリを監視アラートとして設定していきます。
右上の3点リーダーで「Save As..」をクリックします。

アラートとして設定する Scheduled Search の名前を設定して、「schedule this search」に進みます。

Run frequency は「Hourly」くらいが適当でしょう。0より大きいクエリ結果が出てきた場合にアラートを送出するように設定して、Emailの送信先を設定します。

これで監視が完了です。やりたいことができました。

クエリの解説

ここからは、先程の監視のためにつくったクエリの解説になります。
先程のクエリの出力結果の部分まで話を戻します。

クエリを再度詳しく見ていくと、「where」オペレーターは条件検索となります。
最後に出力されていたログより60分以上経過したものを条件としていることが分かります。また、最初に設定した監視したいコレクター名と一致するものを条件としています。

_index=sumologic_volume sizeInBytes _sourceCategory="collector_volume"
| parse regex "\"(?<collector>[^\"]+)\"\:\{\"sizeInBytes\"\:(?<bytes>\d+),\"count\"\:(?<count>\d+)\}" multi
| first(_messagetime) as MostRecent, sum(bytes) as TotalVolumeBytes by collector
| formatDate(fromMillis(MostRecent),"yyyy/MM/dd HH:mm:ss") as MostRecentTime
| toMillis(now()) as currentTime
| formatDate(fromMillis(currentTime),"yyyy/MM/dd HH:mm:ss") as SearchTime
| (currentTime-MostRecent) / 1000 / 60 as mins_since_last_logs
| where mins_since_last_logs >= 60
| fields -mostrecent, currenttime
| format ("%s Has not collected data in the past 60 minutes", collector) as message
| where collector = <collector名>

次に、Messages をクリックして、クエリ文の条件検索でマッチしなかったものも含めた検索結果を見ていきます。
そうすると、「parse」オペレーターでログを解析したフィールドと生のログメッセージを確認することができます。

「parse regex」では、生のログを正規表現でコレクター名に当たるところ「InternalCollector」や「aws-observability-533554244959-533554244959」などを「collector」というフィールドで解析します。同様に「bytes」、「count」部分を解析しています。
フィールドとして解析することで後続のクエリ文で計算や集計等の様々な処理に利用できます。

_index=sumologic_volume sizeInBytes _sourceCategory="collector_volume"
| parse regex "\"(?<collector>[^\"]+)\"\:\{\"sizeInBytes\"\:(?<bytes>\d+),\"count\"\:(?<count>\d+)\}" multi
| first(_messagetime) as MostRecent, sum(bytes) as TotalVolumeBytes by collector
| formatDate(fromMillis(MostRecent),"yyyy/MM/dd HH:mm:ss") as MostRecentTime
| toMillis(now()) as currentTime
| formatDate(fromMillis(currentTime),"yyyy/MM/dd HH:mm:ss") as SearchTime
| (currentTime-MostRecent) / 1000 / 60 as mins_since_last_logs
| where mins_since_last_logs >= 60
| fields -mostrecent, currenttime
| format ("%s Has not collected data in the past 60 minutes", collector) as message
| where collector = <collector名>

その後の集計処理では、直近のメッセージのタイムスタンプを取得して、今の時刻からのどれだけのタイムラグがあるかを調べています。
この集計によって、「where mins_since_last_logs >= 60」により60分以上ログ出力の無いコレクターが無かったかを調べることができます。

_index=sumologic_volume sizeInBytes _sourceCategory="collector_volume"
| parse regex "\"(?<collector>[^\"]+)\"\:\{\"sizeInBytes\"\:(?<bytes>\d+),\"count\"\:(?<count>\d+)\}" multi
| first(_messagetime) as MostRecent, sum(bytes) as TotalVolumeBytes by collector
| formatDate(fromMillis(MostRecent),"yyyy/MM/dd HH:mm:ss") as MostRecentTime
| toMillis(now()) as currentTime
| formatDate(fromMillis(currentTime),"yyyy/MM/dd HH:mm:ss") as SearchTime
| (currentTime-MostRecent) / 1000 / 60 as mins_since_last_logs
| where mins_since_last_logs >= 60
| fields -mostrecent, currenttime
| format ("%s Has not collected data in the past 60 minutes", collector) as message
| where collector = <collector名>

※監視対象にできるのは、通常時コンスタンスにログの取り込みが発生しているものになります。既に停止しているコレクターやしきい値に設定した時間以上（60分など）を超えてログの取り込みが無い状態が発生する対象には指定できません。

その他の観点でログの取り込みを監視する

これまではコレクター毎でログ出力が止まっていないかを見ていましたが、その他にソースカテゴリーで確認したい場合もあるかもしれません。
その場合は、以下のようなクエリに変更すればソースカテゴリー毎に見ることが可能となります。

_index=sumologic_volume sizeInBytes _sourceCategory="sourceCategory_volume"
| parse regex "\"(?<sourceCategory>[^\"]+)\"\:\{\"sizeInBytes\"\:(?<bytes>\d+),\"count\"\:(?<count>\d+)\}" multi
| first(_messagetime) as MostRecent, sum(bytes) as TotalVolumeBytes by sourceCategory
| formatDate(fromMillis(MostRecent),"yyyy/MM/dd HH:mm:ss") as MostRecentTime
| toMillis(now()) as currentTime
| formatDate(fromMillis(currentTime),"yyyy/MM/dd HH:mm:ss") as SearchTime
| (currentTime-MostRecent) / 1000 / 60 as mins_since_last_logs
| where mins_since_last_logs >= 60
| fields -mostrecent, currenttime
| format ("%s Has not collected data in the past 60 minutes", sourceCategory) as message
| where sourceCategory = <sourceCategory名>

また、ソースカテゴリーは一般的に命名規則を持たせて設計することが多いので、に「aws/*」のようなアスタリスクを指定することで複数のソースカテゴリを対象に監視することができます。

まとめ

以上、Sumo Logic のデータの取り込みが正しく行われているかを監視するための Tips でした。本ブログが誰かの一助となれば幸いです。