Splunk の eventstats と streamstats の違いと利用方法をまとめてみた

タイトルの eventstats、streamstats について考えてみます。
Splunkを使っていると、なんかいい感じの統計が出来るって聞いてたけど、なんだったっけ？何が違くてどんな時に使うんだっけ？ってなりがちなコマンドかなと思います。
他のブログ記事でもよく書かれてはいるのですが、自分の備忘録として記事にしようと思いました。

いきなりまとめ

stats - 全てのレコードを統計の対象とする。統計前のデータは表示、参照できない
eventstats - 全てのレコードを統計の対象とする。統計前のデータは表示、参照できる
streamstats - 計算するタイミングまでのレコードを統計の対象とする（window=xx を使って統計対象のレコード数を指定することが可能）。統計前のデータは表示、参照できる

stats は統計するコマンド

stats は count、sum、avg、values などの「stats function」を指定して統計処理を行い、表形式でデータをまとめるコマンドで、一番よく使う統計コマンドといって過言ではないかと思います。

以下のような感じで使います。

| stats count sum<field1> avg<field1> values<field2> by <field3>

https://docs.splunk.com/Documentation/SplunkCloud/latest/SearchReference/stats

一方、eventstats、streamstatsも全く同じ「stats function」を利用することができ、一見すると何が違うのか分からなくなりがちです。

stats と eventstats、streamstats それぞれの違い

それぞれのコマンドの動きの違いについて確認してみます。

※下記からチュートリアルデータがダウンロードできます。サンプルコマンド結果はこちらを利用しています。
https://docs.splunk.com/Documentation/Splunk/latest/SearchTutorial/Systemrequirements#Download_the_tutorial_data_files

statsを使って10行のデータを対象にclientipごとのbytesの合計値を計算してみます。

SPLコマンド:

sourcetype="access_combined_wcookie"
| head 10
| stats sum(bytes) as total_bytes by clientip

結果:

さらに、各レコードのbytesの値も同時に表示したいとします。

SPLコマンド:

sourcetype="access_combined_wcookie"
| head 10
| stats sum(bytes) as total_bytes by clientip
| table bytes total_bytes clientip

結果:

そうすると、bytesの値は参照することはできなくなります。

これがeventstatsを利用すると動作が変わります。

SPLコマンド:

sourcetype="access_combined_wcookie"
| head 10
| eventstats sum(bytes) as total_bytes by clientip
| table bytes total_bytes clientip

結果:

まず、10行分のレコードが返されたことが分かります。
またbytesの結果は後からでも参照したり表示したりすることができます。

次にstreamstatsを使ってみます。

SPLコマンド:

sourcetype="access_combined_wcookie"
| head 10
| streamstats sum(bytes) as total_bytes by clientip
| table bytes total_bytes clientip

結果:

bytesとclientipについては出力結果はstreamstatsと変わりません。
total_bytesが変わっていることが分かりますが、結果を２つのclientipごとで分けて見てみます。

結果:

結果:

上から順番にそれまでのレコードのbytesの合計結果が表示されるようになっていることが分かります。

どんな時に使うか

statsは最も基本的なコマンドで利用用途は分かりやすいので飛ばします。

eventstats

1. 時系列データに対して標準偏差による動的しきい値を算出し外れ値を検出

streamstatsも同じユースケースが可能ですが、以下の部分が少し違います。
タイムピッカーで指定したデータ範囲全体を対象に、動的しきい値を算出します。

以下のクエリでは、アクセスクエリのイベント数を時系列データにして、タイムピッカーで指定したデータの平均値と2標準偏差を用いて、外れ値を算出します。
※タイムピッカーは全期間を指定しています。

sourcetype=access_combined_wcookie
| timechart count span=1h
| eventstats avg(count) as avg stdev(count) as stdev
| eval upper_bound = avg + 2*stdev, lower_bound = avg - 2*stdev
| fields - avg - stdev
| eval outlier = if(count > upper_bound OR count < lower_bound , 1, 0)

標準偏差を使った統計手法は他のデータにもそのまま流用することができるので、ぜひ試してみてください。
2標準偏差や3標準偏差に調整することで、外れ値の厳しさを決めることができます。

2. これまでの利用用途では観測されなかった特異な行動パターンの検出

タイムピッカーで指定したデータ（1ヶ月や3ヶ月など）を対象に、それまでは観測されていなかったが直近（24時間以内など）で観測されたデータを見つけます。

検索の例としては以下のように利用できます。
※タイムピッカーは1ヶ月または3ヶ月などの期間を指定します。任意で調整してください。

sourcetype=access_combined_wcookie
| iplocation clientip
| eventstats earliest(_time) as earliest latest(_time) as latest by Country
| eval maxlatest=now()
| where earliest >= relative_time(maxlatest, "-24h")
| stats count by clientip, Country, City
| table clientip, count, Country, City 
| sort - count

少し分かりづらいので、クエリ文を一行ずつ説明すると、
clientipフィールドでIPアドレスの情報から、緯度や国・都市の地理的情報をとってきます。

| iplocation clientip

国ごとで分類したイベントの最初のイベント発生時間を取得します。
※eventstatsは統計前のデータを参照することができるので、この後のstatsコマンドも実行できます。

| eventstats earliest(_time) as earliest latest(_time) as latest by Country

現在の時刻を算出します。

| eval maxlatest=now()

現在の時刻から24時間以内に初めて観測されたイベントのみに限定します。

| where earliest >= relative_time(maxlatest, "-24h")

24時間以内に発生したイベント数を集計します。

| stats count by clientip, Country, City

出力形式を整形します。

| table clientip, count, Country, City 
| sort - count

このような特異行動パターンを検出する分析方法は、他に以下のような例で応用することができます。
セキュリティ分析においては、非常に興味深いパターンとして監視すべき観点になります。

• 特定のユーザーアカウントでこれまで実行していなかったアクションを検出
• 普段アクセスしないようなユーザーエージェント
• 新しいユーザーアカウントの作成、休眠アカウントの利用を検出

また、注目したい観点を変えればどんなものでも応用ができそうです。

streamstats

1. 時系列データに対して標準偏差による動的しきい値を算出し外れ値を検出

先程のeventstatsと同じユースケースですが、以下の部分に違いがあります。
タイムピッカーで指定したデータ範囲内で、さらに範囲を区切って動的しきい値を算出することができます。

こちらも標準偏差を用いた動的しきい値の計算ですが、streamstatsのwindow=12 -> 12個分のデータを時系列でずらしながらしきい値を計算していきます。
※タイムピッカーは全期間を指定しています。

sourcetype=access_combined_wcookie
| timechart count span=1h
| streamstats window=12 avg(count) as avg stdev(count) as stdev
| eval upper_bound = avg + 2*stdev, lower_bound = avg - 2*stdev
| fields - avg - stdev
| eval outlier = if(count > upper_bound OR count < lower_bound, 1, 0)

しきい値は、時系列の流れに沿って流動的に変化していきます。

2. ２つの異なるタイムレンジのデータ変動性を比較して異常性を検出

前のケースの応用になります。
下記は、同一ユーザーでの時系列上の２つのイベントごとの経過時間と地理的情報を算出して、人間が移動不可能な速さで移動している（アカウントが乗っ取られた）ことを検出します。

sourcetype=aws:cloudtrail
| iplocation src
| where isnotnull(lat) AND isnotnull(lon) 
| streamstats window=2 global=false earliest(lat) as prev_lat, earliest(lon) as prev_lon, earliest(_time) as prev_time, earliest(src) as prev_src, earliest(City) as prev_city, earliest(Country) as prev_country, earliest(app) as prev_app by user 
| where (src != prev_src) 
| eval lat1_r=((lat * 3.14159265358) / 180), lat2_r=((prev_lat * 3.14159265358) / 180), delta=(((prev_lon - lon) * 3.14159265358) / 180), distance=(3959 * acos(((sin(lat1_r) * sin(lat2_r)) + ((cos(lat1_r) * cos(lat2_r)) * cos(delta))))), distance=round(distance,2) 
| fields - lat1_r, lat2_r, long1_r, long2_r, delta 
| eval time_diff=if((('_time' - prev_time) == 0),1,('_time' - prev_time)), speed=round(((distance * 3600) / time_diff),2) 
| where (speed > 500) 
| eval prev_time=strftime(prev_time,"%Y-%m-%d %H:%M:%S")
| table user, src, _time, City, Country, app, prev_src, prev_time, prev_city, prev_country, prev_app, distance, speed

経過時間と地理情報から求められる移動不可能なスピードの算出はこのまま利用することができます。
イベントーソースやフィールド名を変えることで流用することができます。

まとめ

eventstats、streamstatsは統計処理をメモリとして一時的に保存する時に使うことができます。
それぞれのコマンドのユースケースを2つずつ紹介してみましたが、まだまだ他にもありそうですので、気づいた時にアップデートしたいと思います。

参考

eventstats:
https://docs.splunk.com/Documentation/Splunk/latest/SearchReference/Eventstats

streamstats:
https://docs.splunk.com/Documentation/Splunk/latest/SearchReference/Streamstats