Splunkで時系列データを集計する   その２

[その１](https://qiita.com/toshikawa/items/1a8fb0f85ebfd5ed8ac1)でもいろいろやったが、それよりも面倒なやつをやったので忘備録としてまとめます。
#課題
```
[02/18/2020 09:45:15.1318] CAUAJM_I_40244 EVENT: CHANGE_STATUS STATUS: STARTING JOB: CFDW_ADHOC_A_AIMSAS_D_INV_LNITEM_BILLING_CHGS_M MACHINE: XXXX
[02/18/2020 10:47:15.1318] CAUAJM_I_40245 EVENT: CHANGE_STATUS STATUS: STARTING JOB: CFDW_ADHOC_C_AIMSAS_D_INV_LNITEM_BILLING_CHGS_M MACHINE: XXXX
[02/18/2020 10:47:35.1318] CAUAJM_I_40244 EVENT: CHANGE_STATUS STATUS: RUNNING JOB: CFDW_ADHOC_A_AIMSAS_D_INV_LNITEM_BILLING_CHGS_M MACHINE: XXXX
[02/18/2020 10:48:15.1318] CAUAJM_I_40245 EVENT: CHANGE_STATUS STATUS: RUNNING JOB: CFDW_ADHOC_C_AIMSAS_D_INV_LNITEM_BILLING_CHGS_M MACHINE: XXXX
[02/18/2020 18:25:15.1318] CAUAJM_I_40245 EVENT: CHANGE_STATUS STATUS: SUCCESS JOB: CFDW_ADHOC_C_AIMSAS_D_INV_LNITEM_BILLING_CHGS_M MACHINE: XXXX
[02/18/2020 19:25:15.1318] CAUAJM_I_40244 EVENT: CHANGE_STATUS STATUS: SUCCESS JOB: CFDW_ADHOC_A_AIMSAS_D_INV_LNITEM_BILLING_CHGS_M MACHINE: XXXX
```
なログがあり、_STARTING_から_SUCCESS_までの時間の各５分ごとにJOB(例：CFDW_ADHOC_A_AIMSAS_D_INV_LNITEM_BILLING_CHGS_M)数を数え、時系列データで表示したい。

#問題
1.各時間のログがない。
みてのとおり、_STARING_ _RUNNING_ _SUCCESS_の三つの状態のログしかなく、その間の経過時間は集計できる。
しかし、その間の各５分間のログはもちろんない。
2.状態が三つ
二つだと楽。しかし三つ、しかも５分間隔だと同時に動くJOBがある。
3.最後はグラフ表示
![qiita3.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/130729/dd7a7d62-197a-2df7-519b-6eca636b9bf7.png)
こんな形をご所望でした。
#作りました。
```java:timeline2.spl
| makeresults
| eval _raw="[02/18/2020 09:45:15.1318] CAUAJM_I_40244 EVENT: CHANGE_STATUS STATUS: STARTING JOB: CFDW_ADHOC_A_AIMSAS_D_INV_LNITEM_BILLING_CHGS_M MACHINE: XXXX
[02/18/2020 10:47:15.1318] CAUAJM_I_40245 EVENT: CHANGE_STATUS STATUS: STARTING JOB: CFDW_ADHOC_C_AIMSAS_D_INV_LNITEM_BILLING_CHGS_M MACHINE: XXXX
[02/18/2020 10:47:35.1318] CAUAJM_I_40244 EVENT: CHANGE_STATUS STATUS: RUNNING JOB: CFDW_ADHOC_A_AIMSAS_D_INV_LNITEM_BILLING_CHGS_M MACHINE: XXXX
[02/18/2020 10:48:15.1318] CAUAJM_I_40245 EVENT: CHANGE_STATUS STATUS: RUNNING JOB: CFDW_ADHOC_C_AIMSAS_D_INV_LNITEM_BILLING_CHGS_M MACHINE: XXXX
[02/18/2020 18:25:15.1318] CAUAJM_I_40245 EVENT: CHANGE_STATUS STATUS: SUCCESS JOB: CFDW_ADHOC_C_AIMSAS_D_INV_LNITEM_BILLING_CHGS_M MACHINE: XXXX
[02/18/2020 19:25:15.1318] CAUAJM_I_40244 EVENT: CHANGE_STATUS STATUS: SUCCESS JOB: CFDW_ADHOC_A_AIMSAS_D_INV_LNITEM_BILLING_CHGS_M MACHINE: XXXX"
| makemv delim="
" _raw
| stats count by _raw
| rex "\[(?<timestamp>.*)\] (?<session>\S+) .*STATUS: (?<status>\S+) JOB: (?<job>\S+) MACHINE: (?<machine>\S+)"
| table timestamp session status job machine
| eval _time=strptime(timestamp,"%m/%d/%Y %T.%3Q")
| eval status=if(status="STARTING" OR status="RUNNING","RUNNING",status)
| eval job_status=job.":".status
| timechart cont=f span=5m count by job_status
| makecontinuous span=5m _time
| foreach *:RUNNING [ eval <<FIELD>> = nullif('<<FIELD>>','<<MATCHSTR>>:SUCCESS')]
| foreach *:SUCCESS [ eval <<FIELD>> = if(isnull('<<MATCHSTR>>:RUNNING') AND '<<FIELD>>'=0 ,NULL, '<<FIELD>>')]
| reverse
| filldown *:SUCCESS
| reverse
| foreach *:RUNNING [ eval <<FIELD>> = if(isnull('<<FIELD>>') AND '<<MATCHSTR>>:SUCCESS' > 0 , 1, '<<FIELD>>' )]
| filldown *:RUNNING
| foreach *:* [ eval <<MATCHSEG1>> = max('<<MATCHSEG1>>:RUNNING','<<MATCHSEG1>>:SUCCESS')]
| fields - *:*
```
#解説
- `| eval status`や`|eval job_status`のところは２つのフィールドを同時に扱うのはしんどかったのでまとめました。
- `| timechart cont=f` と　`makecontinuous`で ログの時間で各５分間ごとの時間を作成
- `reverse`前の`foreach`は時間によって別なJOBが両方`0`な場合があるので`NULL`にしている。<br>
これは`filldown`を使用する前の準備<br>
あえて`nullif`を使って`NULL`を作ってみました。
- `filldown`は`*(アスタリスク)`が使用できるので、こんなことができます。
- `reverse`を２回やっているのはひっくり返して最後(_SUCCESS_）から`1`を埋めていっている。そうすると２番目(_RUNNING_)の`0`のところで止まってくれる。<br>それで元に戻している。
- 次の`foreach`は最初から２番目までを1で埋める前の下準備でJOBが`2`の時刻以外を`1`や`0`に戻している。
- 最後の`foreach`で２つの状態の値があるので、最終的にJOBの名前に統一してあげている。<br>`eval`で`max`が使えるので(2,1,0)の値の一番大きい値としている。
- お疲れ様でした。

#苦労したところ
1. `timechart cont=f`から`makecontinuous`はやってみたらすんなり出来た
2. `streamstats`で頑張れるかなとも思ったけど、JOB数が増えることに対応出来なさそうだったから早急に諦めました。`stats count(any*) as *`は可能だけど`eval`を使うとなると`*`が使えないので無理ですね。
3. `filldown`を使うのは早々と決めたがJOBが`2`のところで大苦戦。こいつを次の時間は`1`にしなきゃいけない。<br>`| foreach *:RUNNING [ eval <<FIELD>> = if(isnull('<<FIELD>>') AND '<<MATCHSTR>>:SUCCESS' > 0 , 1, '<<FIELD>>' )]`のところの前後で一回値を確かめてみてください。:cry:
4. `timechart`で時間を作っているのでどうしてもJOBの片方は`0 0`の時がある。`eval`でいれてもよかったかもしれなかったけど、あえて`foreach`でJOBが増えても大丈夫にしている。<br>SPLは値を保持出来ないのでこんな感じでやってます。もっとスッキリとやりたかった:sweat:

#やりなおしてみた
```java:timechart2.spl
| makeresults
| eval _raw="[02/18/2020 09:45:15.1318] CAUAJM_I_40244 EVENT: CHANGE_STATUS STATUS: STARTING JOB: CFDW_ADHOC_A_AIMSAS_D_INV_LNITEM_BILLING_CHGS_M MACHINE: XXXX
[02/18/2020 10:47:15.1318] CAUAJM_I_40245 EVENT: CHANGE_STATUS STATUS: STARTING JOB: CFDW_ADHOC_C_AIMSAS_D_INV_LNITEM_BILLING_CHGS_M MACHINE: XXXX
[02/18/2020 10:47:35.1318] CAUAJM_I_40244 EVENT: CHANGE_STATUS STATUS: RUNNING JOB: CFDW_ADHOC_A_AIMSAS_D_INV_LNITEM_BILLING_CHGS_M MACHINE: XXXX
[02/18/2020 10:48:15.1318] CAUAJM_I_40245 EVENT: CHANGE_STATUS STATUS: RUNNING JOB: CFDW_ADHOC_C_AIMSAS_D_INV_LNITEM_BILLING_CHGS_M MACHINE: XXXX
[02/18/2020 18:25:15.1318] CAUAJM_I_40245 EVENT: CHANGE_STATUS STATUS: SUCCESS JOB: CFDW_ADHOC_C_AIMSAS_D_INV_LNITEM_BILLING_CHGS_M MACHINE: XXXX
[02/18/2020 19:25:15.1318] CAUAJM_I_40244 EVENT: CHANGE_STATUS STATUS: SUCCESS JOB: CFDW_ADHOC_A_AIMSAS_D_INV_LNITEM_BILLING_CHGS_M MACHINE: XXXX"
| makemv delim="
" _raw
| stats count by _raw
| rex "\[(?<timestamp>.*)\] (?<session>\S+) .*STATUS: (?<status>\S+) JOB: (?<job>\S+) MACHINE: (?<machine>\S+)"
| table timestamp session status job machine
| eval _time=strptime(timestamp,"%m/%d/%Y %T.%3Q")
| bin span=5min _time
| reverse
| streamstats values(_time) as range  by session
| reverse
| eval start=mvindex(range,0), end=mvindex(range,1)
| streamstats current=f values(start) as start_p1 by job
| eval end=if(start=start_p1,null(),end)
| eval timerange=if(isnull(end),start,mvrange(start,end,300))
| fields _time status job timerange
| mvexpand timerange
| rename timerange as _time
| table _time job status
| timechart span=5m count(status) as count by job
```
すっきり、画面は一緒になったので省略
#解説その２
よくよく考えてみると、[その１](https://qiita.com/toshikawa/items/1a8fb0f85ebfd5ed8ac1)でやった`mvexpand`が使えんじゃね。ということでやってみた。

`bin`で時間を整えて`reverse`と`streamstats`で経過時間を抽出。

やはりjob数が`2`になるところで苦戦:cry:　重複してログが出来てしまう。
結局直前の時間と比較`| eval end=if(start=start_p1,null(),end)`して`| eval timerange=if(isnull(end),start,mvrange(start,end,300))`で時間を作ってあげることで解決した。
こうすることで、重複している時間と`status="SUCCESS"`のところはログが一つしか作らなくなる。

あとは`mvexpand`してあげれば、素直にログができるので、`timechart`で数えておしまい。
:sweat:なんだろう、この素直さは
#まとめ
ログがあれば、数えて終わりだと思います。

無いログを作れるのがSplunkのいいところ。
似たような案件を抱えている方がいましたら、`japanese`のタグをつけて日本語で[Splunk>Answers](https://answers.splunk.com/)に質問してみてください。
サンプルログ必須ですけど、クエリーは作成します。

ダッシュボードは[心の師匠](https://answers.splunk.com/users/299862/niketnilay.html)に聞いてください

-----
#####更新履歴
R2.3.11 やりなおしてみた　を追加