Elasticsearchから複数のイベントのデータを1つのlogstashフィルタを使用して抽出する

Question

複数のイベントに散在するデータを持つElasticSearchにログラインがロードされている、event_idがイベント（行）番号5にあり、event_actionがイベント番号88 event_port情報はイベント番号455にあります。このデータをどのように抽出して、出力が次のようになるのでしょうか。この場合、マルチラインコーデックは機能しません。

{ 
event_id: 1223 
event_action: "socket_open" 
event_port: 76654 
} 

現在、私は、ログファイルはので、私はESからファイルパスを取得することができます持続しています。私はrubyフィルタからシェルスクリプトを実行しようとしましたが、このシェルスクリプトはgrepコマンドを実行し、stdoutデータを次のような新しいイベントに入れます。

input { 
    elasticsearch { 
    hosts => "localhost:9200" 
    index => "my-logs" 
    } 
} 

filter 
{ 

    ruby { 
    code => 'require "open3" 
      file_path = event.get("file_path") 
      cmd = "my_filter.sh -f #{file_path}" 
      stdin, stdout, stderr = Open3.popen3(cmd) 
      event.set("process_result", stdout.read) 
      err = stderr.read 
      if err.to_s.empty? 
       filter_matched(event) 
      else 
       event.set("ext_script_err_msg", err) 
      end' 
     remove_field => ["file_path"] 
    } 
    } 

上記のアプローチでは、私は問題に直面しています。

1）膨大なファイルでgrepを実行すると時間がかかることがあります。ファイルにgrepする必要はありませんか？

2）私の入力プラグイン（上に添付されています）は、インデックスの "ALL"イベントにfile_pathが設定されているElastic Searchのイベントを受け取ります。これにより、my_filter.shが複数回実行されます。 ESから一意のfile_pathを抽出するにはどうすればよいですか？

Answer 1

入力に応じて出力ストリームを構築するために、Elasticsearchを作成しませんでした。 ElasticはnoSQLデータベースで、データは時間の経過とともに（リアルタイムアプローチで）消費される必要があります。つまり、最初にすべてをElasticsearchに保存してからデータを処理する必要があります。あなたの場合は、別のイベントを待ってフローを張っています。

これらのイベントを捕捉してバックグラウンドで処理する必要がある場合は、logstash（入力はnxlog）やpythonスクリプト（logstashのフィルタとして使用）でフィルタリングする前にnxlogのようなものを試すことができます。あなたのケースでは、私はそれを統合するために私のデータを前処理し、それをlogstashに送る