Ingestnode Grokにてarraylistの処理について

harue · March 6, 2020, 1:06am

お世話になります。

ingest nodeのGrok processorにて、配列の値を処理したいのですがエラーになります。

POST /_ingest/pipeline/_simulate
{
  "pipeline": {
    "processors": [
      {
        "kv": {
          "field": "message",
          "field_split": """\s(?![\\\[\]!"#$%&'()*+,-./:;<>?@\^`{|}~\w[^ -~｡-ﾟ] ]+?(\s+|$))""",
          "value_split": "="
        }
      },
      {
        "grok": {
          "field": "request",
          "patterns": [
            "%{GREEDYDATA}://%{USERNAME:requestdomain}/%{GREEDYDATA}"
          ],
          "ignore_missing": true
        }
      },
      {
        "script": {
          "lang": "painless",
          "if": "ctx.requestdomain != null",
          "source": "ctx.requestdomain = '*' + (ctx.requestdomain) + '*' "
        }
      }
    ],
    "on_failure": [
      {
        "set": {
          "field": "_index",
          "value": "failed-index_ddi"
        }
      },
      {
        "set": {
          "field": "error",
          "value": "{{ _ingest.on_failure_message }}"
        }
      }
    ]
  },
  "docs": [
    {
      "_index": "aaa",
      "_id": "id1",
      "_source": {
        "message": "request=http://aaa.com/ request=http://bbb.com/"
      }
    }
  ]
}

実行結果

{
  "docs" : [
    {
      "doc" : {
        "_index" : "failed-index_ddi",
        "_type" : "_doc",
        "_id" : "id1",
        "_source" : {
          "request" : [
            "http://aaa.com/",
            "http://bbb.com/"
          ],
          "message" : "request=http://aaa.com/ request=http://bbb.com/",
          "error" : "field [request] of type [java.util.ArrayList] cannot be cast to [java.lang.String]"
        },
        "_ingest" : {
          "timestamp" : "2020-03-06T00:49:45.079364Z"
        }
      }
    }
  ]
}

■実現したいこと

"request" : ["http://aaa.com/","http://bbb.com/"]の「http://XXX/」のXXXを抽出
↓
XXXの値の前後に"*"を追加
"requestdomain" : ["*aaa*","*bbb*"]

grok processorにて配列処理を実施することは出来ないのでしょうか？
実現する方法について分かりましたら、教えてください。

tsgkdt · March 6, 2020, 5:35am

Grok Processorが配列には対応していないと思いますので、全部Painless Scriptで実行するのはどうでしょうか？

POST /_ingest/pipeline/_simulate
{
  "pipeline": {
    "processors": [
      {
        "kv": {
          "field": "message",
          "field_split": """\s(?![\\\[\]!"#$%&'()*+,-./:;<>?@\^`{|}~\w[^ -~｡-ﾟ] ]+?(\s+|$))""",
          "value_split": "="
        }
      },
      {
        "script": {
          "lang": "painless",
          "source": """
              ctx.requestdomain = []; 
              for (line in ctx.request) {
                def matcher = /^https?:\/{2,}(.*?)\//.matcher(line);
                if (matcher.find()) { 
                  ctx.requestdomain.add('*' + matcher.group(1).trim() + '*'); 
                }
              }
            """
        }
      }
    ],
    "on_failure": [
      {
        "set": {
          "field": "_index",
          "value": "failed-index_ddi"
        }
      },
      {
        "set": {
          "field": "error",
          "value": "{{ _ingest.on_failure_message }}"
        }
      }
    ]
  },
  "docs": [
    {
      "_index": "aaa",
      "_id": "id1",
      "_source": {
        "message": "request=http://aaa.com/ request=http://bbb.com/"
      }
    }
  ]
}

その結果がこちら。

{
  "docs" : [
    {
      "doc" : {
        "_index" : "aaa",
        "_type" : "_doc",
        "_id" : "id1",
        "_source" : {
          "request" : [
            "http://aaa.com/",
            "http://bbb.com/"
          ],
          "requestdomain" : [
            "*aaa.com*",
            "*bbb.com*"
          ],
          "message" : "request=http://aaa.com/ request=http://bbb.com/"
        },
        "_ingest" : {
          "timestamp" : "2020-03-06T05:31:21.550618Z"
        }
      }
    }
  ]
}

およそ実現されたいことができているのではないかと思います。
ただし、これを行うためには、elasticsearch.ymlに以下を記述する必要があります。

script.painless.regex.enabled: true

過去の類似の問い合わせとしては、以下になるかと思います。

Ingest: foreachとgrokを組み合わせた場合の処理について日本語による質問・議論はこちら

知りたいこと Ingest Pipelineでforeachの中でgrokを適用し、結果をarrayとして保存する方法について状況の説明括弧内にログのコード、次にログのメッセージからなる文字列が、カンマ区切りで出力されているテキストがあります。（例） [INFO.0001]あいうえおかきくけこ, [INFO.0002]さしすせそこれを、Ingest Pipelineでこのようにパースしたいと考えています。期待値 { "hoge": ["INFO.0001", "INFO.0002"], "fuga": ["あいうえおかきくけこ", "さしすせそ"] } try まず、splitで配列にし、foreach内でgrokによりコードとメッセージを取るように指定することを考えました。これだと、hoge, fugaのフィールドに対して最後のループのときの値がセットされるだけです。 { "split": { "field": "message", "separator": "," } },…

ご参考になれば幸いです。

harue · March 9, 2020, 7:05am

回答を頂きありがとうございます。
上記Painless Scriptにて実行することができました。

再度の質問で大変恐縮ですが、

"request=http://aaa.com/ request=http://bbb.com"
末尾に/が有と無両方のデータを取込みたいのですが、処理内容がうまくいかず苦戦しています。

requestdomainは以下結果
"requestdomain" : ["*aaa.com*","*bbb.com*" ]

上記Painless Scriptにて、取り込むことは可能でしょうか？

お手数ですが再度回答頂けますと幸いです。

tsgkdt · March 9, 2020, 8:46am

Elastic Stackというよりは正規表現の世界の話になりそうなところですが、
このようにしてみてら、希望する結果になりませんでしょうか？

先の内容と変わっているところは matcherの行です。

POST /_ingest/pipeline/_simulate
{
  "pipeline": {
    "processors": [
      {
        "kv": {
          "field": "message",
          "field_split": """\s(?![\\\[\]!"#$%&'()*+,-./:;<>?@\^`{|}~\w[^ -~｡-ﾟ] ]+?(\s+|$))""",
          "value_split": "="
        }
      },
      {
        "script": {
          "lang": "painless",
          "source": """
              ctx.requestdomain = []; 
              for (line in ctx.request) {
                def matcher = /^https?:\/{2,}(.*?)(\/|$)/.matcher(line);
                if (matcher.find()) { 
                  ctx.requestdomain.add('*' + matcher.group(1).trim() + '*'); 
                }
              }
            """
        }
      }
    ],
    "on_failure": [
      {
        "set": {
          "field": "_index",
          "value": "failed-index_ddi"
        }
      },
      {
        "set": {
          "field": "error",
          "value": "{{ _ingest.on_failure_message }}"
        }
      }
    ]
  },
  "docs": [
    {
      "_index": "bbb",
      "_id": "id1",
      "_source": {
        "message": "request=http://ccc.com/ request=http://ddd.com request=http://eee.co.jp/"
      }
    }
    
  ]
}

実行結果がこちらになります。

{
  "docs" : [
    {
      "doc" : {
        "_index" : "bbb",
        "_type" : "_doc",
        "_id" : "id1",
        "_source" : {
          "request" : [
            "http://ccc.com/",
            "http://ddd.com",
            "http://eee.co.jp/"
          ],
          "requestdomain" : [
            "*ccc.com*",
            "*ddd.com*",
            "*eee.co.jp*"
          ],
          "message" : "request=http://ccc.com/ request=http://ddd.com request=http://eee.co.jp/"
        },
        "_ingest" : {
          "timestamp" : "2020-03-09T08:44:38.592314Z"
        }
      }
    }
  ]
}

動作確認は、7.6.1の環境で行っております。
ご参考になれば幸いです。

harue · March 10, 2020, 5:56am

回答頂きありがとうございます。
大変助かりました。

正規表現の範囲の質問をしてしまい申し訳ありませんでした。
次回からはelastic分野以外の質問は実施しないように気を付けます。

system · April 7, 2020, 6:08am

This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.

Topic		Replies	Views
Ingest node GrokにてArraylistの値を受け取りエラー日本語による質問・議論はこちら	5	838	October 28, 2019
Painlessでの値の取得方法について日本語による質問・議論はこちら	3	1127	April 8, 2020
Ingest: foreachとgrokを組み合わせた場合の処理について日本語による質問・議論はこちら	3	2155	October 31, 2017
【ingest node】JSON processor 配列のパース処理に関して日本語による質問・議論はこちら	2	552	October 8, 2020
Ingest node ドットを含むフィールド名の取得方法について日本語による質問・議論はこちら	3	552	December 30, 2019

Ingestnode Grokにてarraylistの処理について

Related topics