がどのようにKVフィルタ

Question

での解析後Logstashのデータ形式を検証するために、私はのようなメッセージがログに記録があります。

2017-01-06 19:27:53,893 INFO [[ACTIVE] ExecuteThread: '10' for queue: 'weblogic.kernel.Default (self-tuning)'] com.symantec.olp.sas.log.SummaryLogAspect - {country=JP, svid=182, typ=sassum, vid=1120, xffmode=0, initallocdate=2014-12-15T01:08:24Z, xffip=222.230.107.165, seatcnt=1, plgid=2, api=941753644, oslocale=JPN, fng=CJ6FRE1208VMNRQG, req=/228228131/28RXAAPB1DqJj/RSLHL940/EMBXtu+/f+/Zeb/KV1Q/DTXZBFC94ZE5AOmz/mDCqB7zJOARDQO/166180202502162557303662649078783407201612&D09DEEFB7E78065D?NAM=SFlBUy1WQTE2&MFN=VkFJTyBDb3Jwb3JhdGlvbg==&MFM=VkpQMTEx&OLA=JPN&OLO=JPN, llmv=470, oslang=JPN, ctok=166180202502162557303662649078783407201612, resptime=119, epid=70D3B811A994477F957A90985109BE9D, campnid=0, remip=222.230.107.165, lictype=SOS, dbepid=70D3B811A994477F957A90985109BE9D, cid=nav1sasapppex02.msp.symantec.com1481215212435, status=10002, siid=240, skum=21356539, skup=010, psn=O749UPCN8KSY, cip=84.100.138.144, mname=VAIO Corporation, puid=1199, skuf=01100470, st=1481765738387, prbid=5967, mmodel=VJP111, clang=EN, pnfi=1120, cprbid=745, cpmv=7428, euip=222.230.107.165, prcdline=2, dvnm=HYAS-VA16, remdays=0, seatid=ah00s8CIdqUQyW2V, sasvid=106, xlsid=3730, baseactkey=186635290403122706518307794, coupon=651218, translogid=75033f05-9cf2-48e2-b924-fc2441d11d33} 
2017-01-06 19:28:03,894 INFO [[ACTIVE] ExecuteThread: '10' for queue: 'weblogic.kernel.Default (self-tuning)'] com.symantec.olp.sas.log.SummaryLogAspect - {country=JP, svid=182, typ=sassum, vid=1120, xffmode=0, initallocdate=2014-12-15T01:08:24Z, xffip=222.230.107.165, seatcnt=1, plgid=2, api=228228131, oslocale=JPN, fng=1TA6U8RVL0JQXA0N, req=/228228131/28RXAAPB1DqJj/RSLHL940/EMBXtu+/f+/Zeb/KV1Q/DTXZBFC94ZE5AOmz/mDCqB7zJOARDQO/166180202502162557303662649078783407201612&D09DEEFB7E78065D?NAM=SFlBUy1WQTE2&MFN=VkFJTyBDb3Jwb3JhdGlvbg==&MFM=VkpQMTEx&OLA=JPN&OLO=JPN, lpmv=470, oslang=JPN, ctok=166180202502162557303662649078783407201612, resptime=119, epid=70D3B811A994477F957A90985109BE9D, campnid=0, remip=222.230.107.165, lictype=SOS, dbepid=70D3B811A994477F957A90985109BE9D, cid=nav1sasapppex02.msp.symantec.com1481215212435, status=0000, siid=240, skum=21356539, skup=010, psn=28MHHH2VPR4T, cip=222.230.107.165, mname=VAIO Corporation, puid=1199, skuf=01100470, st=1481765738387, prbid=5967, mmodel=VJP111, clang=EN, pnfi=1120, cprbid=745, cpmv=1027, euip=222.230.107.165, prcdline=2, dvnm=HYAS-VA16, remdays=0, seatid=StrlisGXA4yAt1ad, sasvid=130, xlsid=2820, baseactkey=028200017462383754273799438, coupon=123456, translogid=72df4536-6038-4d1c-b213-d0ff5c3c20fb} 

が、私はこれらの照合するには、以下のGROKパターンを使用します。

(?m)%{TIMESTAMP_ISO8601:timestamp} %{LOGLEVEL:severity} \[%{GREEDYDATA:thread}\] %{JAVACLASS:className} - \{%{GREEDYDATA:logmsg}\} 

ポストことを、私はKVフィルタを使用してlogmsgフィールドのフィールドを分割し、興味のあるフィールドのみを含めます。私の質問は、私に興味のあるフィールドのフォーマットをどのように検証できますか？私が言及する必要がある1つの事はある - ログはLOGMSGのフィールドの異なる数が含まれています、私はGREEDYDATA

を使用していた理由を次のように私のLogstash.confがあるということ：

input { 
    kafka { 
    bootstrap_servers => "brokers_list" 
    topics => ["transaction-log"] 
    codec => "json" 
    } 
} 

filter { 
     grok { 
      match => [ "message", "(?m)%{TIMESTAMP_ISO8601:timestamp} %{LOGLEVEL:severity} \[%{GREEDYDATA:thread}\] %{JAVACLASS:className} - \{%{GREEDYDATA:logmsg}\}" ] 
      #overwrite => [ "message" ] 
     } 

     if "_grokparsefailure" not in [tags] { 
      kv { 
       field_split => ", " 
       source => "logmsg" 
       include_keys => ["api", "fng", "status", "cip", "cpmv", "translogid", "coupon", "baseactkey", "xlsid", "sasvid", "seatid", "srcHostname", "serverId" ] 
       allow_duplicate_values => false 
       remove_field => [ "message", "kafka.*", "logmsg"] 
      } 
     } 

     if [api] != "228228131" { 
      mutate { add_tag => "_grokparsefailure" } 
     } 

     date { # use timestamp from the log 
      "match" => [ "timestamp", "YYYY-MM-dd HH:mm:ss,SSS" ] 
      target => "@timestamp" 
     } 

     mutate { 
      remove_field => [ "timestamp" ] # remove unused stuff 
     } 
    } 

output { 
    if "_grokparsefailure" not in [tags] {  
    kafka { 
     topic_id => "invalid topic" 
     bootstrap_servers => "brokers_list" 
     codec => json {}  
    }  
    } else {  
    kafka { 
     topic_id => "valid topic" 
     bootstrap_servers => "brokers_list" 
     codec => json { }  
    }  
    } 
} 

をKVフィルタを解析した後、私はapiフィールドの値を調べて、それが228228131と等しくなければ_grokparsefailureタグを追加して、さらに処理しません。

クライアントIPであるcipのようなInclude_keysに記載されているフィールドのフォーマットを検証できるようにしたいですか？これらのフィールドのデータ形式を検証するにはどうすればよいですか？私のログには異なる数のフィールドが含まれているので、grokレベルでは私は検証できません。 KVで解析した後で、私はそれらのフィールドを取得し、それらを検証することができます。検証では、ESインデックスで定義されている型に準拠していることを検証します。これは、彼らが準拠していない場合、私はそれらをカフカの無効な話題に送りたいからです。

ルビーフィルタを使用して検証する必要がありますか？もしそうなら、サンプルをくれますか？または、KV解析後にイベントを再構築し、その新たに作成されたイベントでgrokを再度使用する必要があります。

これらを示すサンプルがありがたいです。

Answer 1

具体的な例では、助けているだろうが、あなたは正規表現で多くのことを確認することができます。

if [myField] =~ /^[0-9]+$/ { 
    # it contains a number 
} 

またはこのような何か：

if [myField] =~ /^[a-z]+$/ { 
    # it contains lower lowercase letters 
}