Logstash：利用來自靜態csv文件的數據在日誌文件中豐富事件

Question

摘要： 在logstash中使用過濾器，它將從事件字段中讀取值，在外部文件（例如csv）中查找該值並從外部文件比賽。使用來自外部文件的值作爲事件中的額外字段添加。

更多信息： 我有一個事件的日誌文件。事件是這樣的：

{"@timestamp":"2014-06-18T11:52:45.370636+02:00","location":{"MainId":3,"SubId":"5"},"EndRequest":{"Duration":{"Main":0,"Page":6720}}} 

我有這樣一個靜態的CSV文件：

1,left 
2,right 
3,top 

當logstash被處理的事件中，我希望能夠使用過濾器會檢查MainId的價值（例如event = 3），並在csv文件中找到這個值。如果找到，那麼該事件必須得到一個標籤：「top」。

這是一種類似於過濾器「GeoIP」的方式。該事件具有字段值，匹配「數據庫」中的值並返回可添加到事件中的值。

我無法找到可以在上述過程中使用的當前過濾器。我需要自己製作一個自定義過濾器嗎？如果有的話，有人可以提示如何解決這個問題？

Answer 1

我從來沒有見過它寫一個插件，所以我繼續寫了一個非常基本的一個：

# encoding: utf-8 
require "logstash/filters/base" 
require "logstash/namespace" 
require "csv" 

# The cvslookup filter allows you to add fields to an event 
# base on a csv file 

class LogStash::Filters::CSVLookup < LogStash::Filters::Base 
    config_name "csvlookup" 
    milestone 1 

    # Example: 
    # 
    #  filter { 
    #  csvlookup { 
    #  file => 'key_value.csv' 
    #  key_col => 1 
    #  value_col => 2 
    #  default => 'some_value' 
    #   map_field => { "from_field" => "to_field" } 
    #  } 
    #  } 
    # 
    # the default is used if the key_col's value is not present in the CSV file 

    config :file, :validate => :string, :required => true 
    config :key_col, :validate => :number, :default => 1, :required => false 
    config :value_col, :validate => :number, :default => 2, :required => false 
    config :default, :validate => :string, :required => false 
    config :map_field, :validate => :hash, :required => true 

    public 
    def register 
    @lookup = Hash.new 

    CSV.foreach(@file) do |row| 
     @lookup[row[@key_col - 1]] = row[@value_col - 1] 
    end 
    #puts @lookup.inspect 
    end # def register 

    public 
    def filter(event) 
    return unless filter?(event) 

    @map_field.each do |src_field,dest_field| 
     looked_up_val = @lookup[event[src_field].to_s] 
     if looked_up_val.nil? 
      if !@default.nil? 
      event[dest_field] = @default 
      end 
     else 
     if event[dest_field].nil? 
      event[dest_field] = looked_up_val 
     elsif !event[dest_field].is_a?(Array) 
      event[dest_field] = [ event[dest_field], looked_up_val ] 
     else 
      event[dest_field].push(looked_up_val) 
     end 
     end 
    end 
    end # def filter 
end # class LogStash::Filters::CSVLookup 

有可能在其上做進一步的工作 - 例如，如果src_field是一個數組，它可以迭代它，但它應該像你的情況那樣工作。

Answer 2

有翻譯過濾器。

取而代之的是CSV的，你有一個YAML文件，併爲單鍵值對，這應該是一個簡單的SED YAML轉換

在寫作時

最新文檔：http://logstash.net/docs/1.4.2/filters/translate