Watcherで特定のフィールドの値ごとに集計して件数超過で通知する方法

[a-naga]

ElasticsearchにIPフィールドを含むログデータがindexされていく状態で、
以下のようなWatcher定義を検討しています。

過去10分間のログを対象に、IPフィールドのIPアドレス毎に同一IPアドレスの件数を集計し、
5件以上出ている全てのIPアドレスのドキュメント内容をメールで通知する

Watcherでこのような条件が可能かどうか、良い方法があればご教示いただきたいです。

[Koji_Kawamura]

@a-naga さん、

Watcher の Create threshold alert で以下のような Watch を作成するのが簡単だと思います。必要に応じて、画面右下の Show request リンクを押して表示される JSON をコピーして、それをベースに Advanced watch を作成するとカスタマイズできます。

メール本文には {{ctx.payload.results}} を使うと 5回以上出現した IP の結果が取れます。

ここで問題なのは、ctx.payload.results は HashMap になっているので、そのまま出すと

{0={value=5, key=192.168.10.1}, 1={value=5, key=192.168.10.3}}

となります。また、Watcher ではまだ Mustache を使っていて、 Handlebar の {{#each xxx}}{{this}}{{/each}}が使えません。本文で IP のリストをループしてメール本文を出力したい場合は results を単純なリストにしてあげたほうがよさそうです。そうすれば {{#ctx.payload.results}}{{this}}{{/ctx.payload.results}} のように出力できるようになります。

Threshold alert 作成画面の右下にある show request で表示される JSON をコピーして、これをベースに Advanced alert を作るとよいです。

修正が必要なのは transform 部分ですね。

  "transform": {
    "script": {
      "source": "HashMap result = new HashMap(); ArrayList arr = ctx.payload.aggregations.bucketAgg.buckets; ArrayList filteredHits = new ArrayList(); for (int i = 0; i < arr.length; i++) { HashMap filteredHit = new HashMap(); filteredHit.key = arr[i].key; filteredHit.value = arr[i].doc_count; if (filteredHit.value >= params.threshold) { filteredHits.add(filteredHit); } } result.results = filteredHits; return result;",
      "params": {
        "threshold": 5
      }
    }
  }

単純に IP アドレスのリストを返すなら、以下のようになると思われます。(実際に動かして試してないので構文エラーがあるかもです)

  "transform": {
    "script": {
      "source": "ArrayList arr = ctx.payload.aggregations.bucketAgg.buckets; ArrayList filteredHits = new ArrayList(); for (int i = 0; i < arr.length; i++) { if (filteredHit.value >= params.threshold) { filteredHits.add(arr[i].key); } } result.results = filteredHits; return result;",
      "params": {
        "threshold": 5
      }
    }
  }

以上、ご参考までに。

[a-naga]

遅くなりましたが、ご回答ありがとうございます。
ご教示いただいた内容を参考に、件数超過したIPアドレスのリストを取得することができました。

追加で質問させていただきたいのですが、例えば
「xx.xxx.xx.100」のIPアドレスフィールドを持つドキュメントが5件、
「xx.xxx.xx.101」のIPアドレスフィールドを持つドキュメントが5件、
「xx.xxx.xx.102」のIPアドレスフィールドを持つドキュメントが3件あったとして、
IPアドレス毎に5件を超えているか判定して、「xx.xxx.xx.100」と「xx.xxx.xx.101」の
IPアドレス以外のフィールドも含めた10件のドキュメント全内容を取得したいのですがそれは可能でしょうか。