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"
 # }
}