ログを4つのログに分割してElasticsearchに送る

下記のようなログをlogstashで整形してElasticSearchに送りたいのですが可能でしょうか?
<ログのフォーマット>
日付:ログ種別△ログ1△△ログ2△△ログ3△△ログ4
※△はタブです。

ログ1、ログ2、ログ3、ログ4とそれぞれ別のログとして
elasticsearchに送りたいです。

こんにちわ。

LogstashのGrokとMutateとSplitのフィルターを組み合わせるとできるのではと思います。

これを確認するために、簡易に標準入力に文字列を入れて確認してみる設定ファイルを以下に示します。

input {
    stdin{}
}

filter {
   
    grok {
        # 日付:ログ種別△ログ1△△ログ2△△ログ3△△ログ4
        #  △はタブとのこと
        match => { 
            # 文字列からログ1~4の部分をmultilogとして取り出す
            "message" => "%{TIMESTAMP_ISO8601:log.date}:%{WORD:log.type}   %{GREEDYDATA:multilog}\r" 
        }
    }

    mutate {
        # 取得したログ1~4の部分を、arrayにする
        split => { "multilog" => "     " }
    }
   # arrayにしたログ1~4をそれぞれ違うログとなるように分割する
    split {
        field => "multilog"
    }
}
 
output {
    stdout {}
}

標準入力で以下の文字列を入力します。

2019-09-16 13:20:10:hogelog   aaa     bbb     ccc     ddd

とすると、標準出力で以下のように出ます。
1行の標準入力に対して、4つの出力がされていること、
multilogのところをみると、aaa, bbb, ccc, dddとそれぞれの値が設定されていることが確認できるかと思います。

{
      "log.date" => "2019-09-16 13:20:10",
          "host" => "XXX-YYY-ZZZZZ",
      "log.type" => "hogelog",
       "message" => "2019-09-16 13:20:10:hogelog   aaa     bbb     ccc     ddd\r",
      "multilog" => "aaa",
    "@timestamp" => 2019-11-07T06:37:12.259Z,
      "@version" => "1"
}
{
      "log.date" => "2019-09-16 13:20:10",
          "host" => "XXX-YYY-ZZZZZ",
      "log.type" => "hogelog",
       "message" => "2019-09-16 13:20:10:hogelog   aaa     bbb     ccc     ddd\r",
      "multilog" => "bbb",
    "@timestamp" => 2019-11-07T06:37:12.259Z,
      "@version" => "1"
}
{
      "log.date" => "2019-09-16 13:20:10",
          "host" => "XXX-YYY-ZZZZZ",
      "log.type" => "hogelog",
       "message" => "2019-09-16 13:20:10:hogelog   aaa     bbb     ccc     ddd\r",
      "multilog" => "ccc",
    "@timestamp" => 2019-11-07T06:37:12.259Z,
      "@version" => "1"
}
{
      "log.date" => "2019-09-16 13:20:10",
          "host" => "XXX-YYY-ZZZZZ",
      "log.type" => "hogelog",
       "message" => "2019-09-16 13:20:10:hogelog   aaa     bbb     ccc     ddd\r",
      "multilog" => "ddd",
    "@timestamp" => 2019-11-07T06:37:12.259Z,
      "@version" => "1"
}

お試しください。

ご回答ありがとうございました。
確かにこの方法だとログを分割することはできそうです。

追加で質問となり申し訳ないのですが、この分割したログの中身をさらにフィルタしたいのですが、可能なのでしょうか?
例えば、ログ1は「a△a△a」ログ2は「b△b△b」のような内容でタブ区切りになっています。これをフィルタしたいと考えています。
いろいろ試してみているのですが、具体的な方法がわからないのが現状です。

どこまで試されたのか書かれていないので、既にお試しの方法かもしれません・・・

「a△a△a」のような文字をフィルタしてみたい、とのことですが、どのような結果になることを望んでいらっしゃいますか?
この部分がないため、的外れになるかもしれませんがご容赦ください。

["a1", "a2", "a3"]のような配列にしたいのでしょうか?
であれば、再度splitすることで対応できそうです。

    #-------------------中略--------
    mutate {
        split => { "multilog" => "     " }
    }
    split {
        field => "multilog"
    }

    mutate {
        split => { "multilog" => "   " }  # ←今度はタブ1つ分で区切り文字としておく
    }

あるいは、「a△a△a」が「aaa△123△456789」のような場合、次のようにしたいのでしょうか?

    "log_contents" : {
         "first_log" : "aaa",
        "second_log" : "123",
         "thrid_log" :"4566789"
    }

このような場合であっても、Splitのあとに同様にGrokを書いていただければ実現できるかと思います。

    mutate {
        split => { "multilog" => "     " }
    }
    split {
        field => "multilog"
    }

    grok {
        match => { "multilog" => "%{WORD:[log_contents][first_log]}   %{WORD:[log_contents][second_log]}   %{WORD:[log_contents][thrid_log]}" }
    }

grokフィルターの場合は、文字列の形式が不定である場合は困難があります。
(上の例でいえば、タブ区切りで3つの要素という形式に限定されていますね)

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

ありがとうございます。
ご指摘の通り。

  • "log_contents" : {
    
  •      "first_log" : "aaa",
    
  •     "second_log" : "123",
    
  •      "thrid_log" :"4566789"
    
  • }
    

のように実現したかったです。
ご指摘頂いたSplitのあとにGrokを書く方法を試していたのですが、
想定した結果にならず、"_grokparsefailure"と表示されてしまいました。

ログは下記になります。
2019-09-16T13:20:22.789Z:hogelog△△a△a△a△△b△b△b△△c△c△c

grokparsefailureが出るとのことですが、Grokの書式があっているかどうか確認いただくのが良いと思います。

そのときには、KibanaのDevTools以下にあるGrok Debuggerや、こちらが役立つかと思います。

Grok Patternのところには、タブ区切りで3つの要素が並んでいる形式を想定して以下のように書いています。

%{WORD:[log_contents][first_log]}\t%{WORD:[log_contents][second_log]}\t%{WORD:[log_contents][thrid_log]}

Simulateボタンを押すと、結果がペインに表示されます。(マッチしない場合はProvided Grok patterns do not match data in the inputというエラーが出る)

今回はGrokパターンとして1つだけ、タブ区切りで3つに分けられる文字列があったとき、というものしか指定していません。
このパターンにマッチしない場合は、grokparsefailureになります。
他にもパターンがあるかどうか、もあわせて確認いただければと思います。

ツールで確認したところパターンの記述が誤っていました。
大変参考になりました。ありがとうございました。