【ingest node】dissect processorとKV processorを1度に処理できるか

[harue]

お世話になっております。

Ingest nodeについてご質問です。

サンプルログ
value1|value2|key3=value3 key4=value4

上記ログは、1つのログに形式の異なるデータが存在します。

このログを
"value1|value2|箇所は、Dissect Proccesorにて下記の方法で定義

dissect {
   mapping => {
     "message" => "%{key1}|%{key2}"
    }
  }

key3=value3 key4=value4箇所は、KV Proccesorにて下記の方法で定義

{
  "kv": {
    "field": "message",
    "field_split": " ",
    "value_split": "="
  }
}

したいと考えております。

インデックスデータは1つのインデックスに以下のように登録したいです。

field名 value
key1     value1
key2     value2
key3     value3
key4     value4

ご質問
上記2つのprocessorを1つのpipeline上に定義する方法について教えて下さい。
1つのpipeline上では実現不可の場合、pipelineをそれぞれに分けて実現する方法があれば教えてください。

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

[tsgkdt]

1度で定義する、のイメージがよく分からなかったのですが、こういうことです？

Ingest Nodeと書いてあったので、Pipeline processorのsimulateを貼っておきます。

POST _ingest/pipeline/_simulate
{
  "pipeline": {
    "processors": [
      {
        "dissect": {
          "field": "message",
          "pattern": "%{key1}|%{key2}|%{other}",
          "ignore_missing": true
        }
      },
      {
        "kv": {
          "field": "other",
          "field_split": " ",
          "value_split": "="
        }
      },
      {
        "remove": {
          "field": "other"
        }
      }
    ]
  },
  "docs": [
    {
      "_index": "aaa",
      "_id": "1",
      "_source": {
        "message": "value1|value2|key3=value3 key4=value4"
      }
    }
  ]
}

得られる結果がこちら
key1, 2, 3, 4と揃っているように見えますがどうでしょうか？

{
  "docs" : [
    {
      "doc" : {
        "_index" : "aaa",
        "_type" : "_doc",
        "_id" : "1",
        "_source" : {
          "key1" : "value1",
          "key2" : "value2",
          "key3" : "value3",
          "key4" : "value4",
          "message" : "value1|value2|key3=value3 key4=value4"
        },
        "_ingest" : {
          "timestamp" : "2019-12-16T14:26:14.335988Z"
        }
      }
    }
  ]
}

[harue]

早急に回答頂きありがとうございます。
上記パイプラインで想定の値を取得することが出来ました。

追加のご質問で恐縮ですが

サンプルログ(valueにスペースが存在するデータ)
key3=value3 key4=val ue4 key5=value5 key6=val ue6

kv processorで以下を実行すると

POST _ingest/pipeline/_simulate
{
  "pipeline": {
    "processors": [
      {
        "kv": {
          "field": "message",
          "field_split": " ",
          "value_split": "="
        }
      },
      {
        "remove": {
          "field": "message"
        }
      }
    ]
  },
  "docs": [
    {
      "_index": "aaa",
      "_id": "1",
      "_source": {
        "message": "key3=value3 key4=val ue4  key5=value5  key6=val ue6"
      }
    }
  ]
}

以下のエラーが返ってきます。

  {
    "type" : "exception",
    "reason" : "java.lang.IllegalArgumentException: java.lang.IllegalArgumentException: field [message] does not contain value_split [=]",
    "header" : {
      "processor_type" : "kv"
    }

インデックスデータは下記の通りに登録したいです。

field名 value
key3     value3
key4     val ue4
key5     value5
key6     val ue6

上記データをingest nodeで処理する方法はありますでしょうか。
elastic discussの投稿を検索しましたが、回答が見つからずご教示頂けますと幸いです。

How to set Kv filter plugin to keep spaces in value

Hi everyone : I am new to learn ELK, then I'm challenges in using kv filter now. Let me make it brief, the following is one part of my dataset: s1Label=Rule cs2Label=URL Category cs3Label=Virtual System then my configure is filter { kv{ field_split=>" " value_split=>"=" } the result I got was "s1Label":"Rule" "cs2Label":"URL" "cs3Label":"Virtual" How can I get the values including space and set field_split=>" " at same time ? Or is there any other filter to do that …

[tsgkdt]

KV Ingest Processor splitting on whitespace in message

こちらを参考にfield_splitを指定してみたところ、期待する結果になりそうですがどうでしょうか？

POST _ingest/pipeline/_simulate
{
  "pipeline": {
    "processors": [
      {
        "dissect": {
          "field": "message",
          "pattern": "%{key1}|%{key2}|%{other}",
          "ignore_missing": true
        }
      },
      {
        "kv": {
          "field": "other",
          "field_split" : """\s(?![-_,:()\w\" ]+?(\s+|$))""",
          "value_split": "="
        }
      },
      {
        "remove": {
          "field": "other"
        }
      }
    ]
  },
  "docs": [
    {
      "_index": "aaa",
      "_id": "1",
      "_source": {
        "message": "value1|value2|key3=value3 key4=val ue4 key5=value5 key6=val ue6"
      }
    }
  ]
}

結果がこちら

{
  "docs" : [
    {
      "doc" : {
        "_index" : "aaa",
        "_type" : "_doc",
        "_id" : "1",
        "_source" : {
          "key1" : "value1",
          "key2" : "value2",
          "key5" : "value5",
          "key6" : "val ue6",
          "key3" : "value3",
          "key4" : "val ue4",
          "message" : "value1|value2|key3=value3 key4=val ue4 key5=value5 key6=val ue6"
        },
        "_ingest" : {
          "timestamp" : "2019-12-16T16:09:53.719382Z"
        }
      }
    }
  ]
}

ご確認いただけますか？

[harue]

早急に回答頂きありがとうございます。

上記の記載方法で実現できました。
大変助かりました。