Logstash 父子關系 配置

最近在使用Lostash的過程中遇到了一個問題：在一個log文件里包含兩類數據，而且兩類數據之間存在父子關系，那如何使用lostash的configuration實現這個需求呢

思路：

1. 首先定義父事件的pattern，因為子事件不匹配父pattern，所以logstash會自動為子事件添加_grokparesefailure 標簽。通過該標簽即可知道當前事件是父事件還是子事件 
2. 使用filter->ruby生成document_id，並把它放到ruby全局變量中 ，這樣子事件就可以訪問到父事件的document_id
3. 同時為父事件和子事件添加一個字段例如doc_id用來存放步驟二中生成的document_id，單獨為子事件添加一個字段例如parent_id，用來存儲父事件的document_id。

在此要感謝elastic官方論壇的一個帖子:"keep global variable in logstash"，它讓我知道了如何使用filter->ruby來實現全局變量。

以下是logstash的完整配置 

input {
  beats {
    port => 5044
  }
}

filter { 
  # remove the empty lines
  if [message] =~ /^\s*$/ {
    drop { }
  }
  # define parent event pattern
  grok {
    match => {"message" => "%{DATESTAMP:EventTime},%{NUMBER:Mil:INT} %{WORD:Type} %{GREEDYDATA:Item} %{GREEDYDATA:RIC} %{GREEDYDATA:Detail} %{GREEDYDATA:Category}"}
  }
  # children events
  if "_grokparsefailure" in [tags] {
    grok {
      match => {"message" => "\<%{NUMBER:FID:INT}\>,%{GREEDYDATA:FName},%{WORD:FType},%{GREEDYDATA:FValue}"}
      add_field => {"DocID" => '' "ParentID" => ''}
      add_tag => ["%{FType}"]
      remove_tag => ["_grokparsefailure"]
    }
    ruby {
      code => "require 'digest/md5';
      event['ParentID'] = @@parentid;
      event['DocID'] = Digest::MD5.hexdigest(@@parentdate+event['FID'])"
    }
  }
  else{
    mutate {
      add_field => {"DocID" => ''}
      add_tag => ["parent"]
    }
    # define a global variable to keep the parent id
    # must set the default value for the variables in ruby -> init block, or it will raise exception
    ruby {
      init => "@@parentid = '';@@parentdate=''"
      code => "require 'digest/md5';
      @@parentid = Digest::MD5.hexdigest(event['EventTime']+event['Mil']);
      event['DocID'] = @@parentid;
      @@parentdate = event['EventTime']+event['Mil']"
    }
  }
  #remove the redundant fields created by filebeat. you can ignore it if you don't use filebeat as shipper 
  mutate {
    remove_field => ["[beat][hostname]","[beat][name]","count","fields","input_type","offset","type","beat","@version"]
  }  
}

output {
  elasticsearch {
    hosts => ["localhost:9200"]
    index => "%{[@metadata][beat]}-%{+YYYY.MM.dd}"
    #set the document_id
    document_id => %{"DocID"}
    document_type => "%{[@metadata][type]}"
    #template => "/appserver/ELK/logstash-2.3.4/conf/template_tolreport.json"
    #template_name =>"template_tolreport"
    #template_overwrite => true
  }
 # file {
 #   path => "./test-%{+YYYY-MM-dd}.txt"
 # }
}

英語好的同學可以參考我在elastic 的官方論壇中發的帖子：https://discuss.elastic.co/t/logstash-parent-child-event-configuration/58117