Как фильтровать данные JSON из файла log4j с помощью logstash?

Question

У меня есть файл журнала, такой как следующее.

2014-12-24 09:41:29,383 INFO c.t.t.a.c.LoginController.getCSRFToken [http-bio-8080-exec-4] in getCSRFToken 
2014-12-24 09:41:29,383 DEBUG c.t.t.a.c.LoginController.getCSRFToken [http-bio-8080-exec-4] CSRFToken set successfully. 
2014-12-24 09:44:26,607 INFO c.t.t.a.c.LoginController.getCSRFToken [http-bio-8080-exec-8] in getCSRFToken 
2014-12-24 09:44:26,609 DEBUG c.t.t.a.c.LoginController.getCSRFToken [http-bio-8080-exec-8] CSRFToken set successfully. 
2014-12-26 09:55:28,399 INFO c.t.t.a.c.LoginController.getCSRFToken [http-bio-8080-exec-9] in getCSRFToken 
2014-12-26 09:55:28,401 DEBUG c.t.t.a.c.LoginController.getCSRFToken [http-bio-8080-exec-9] CSRFToken set successfully. 
2014-12-26 11:10:32,135 INFO c.t.t.a.c.LoginController.getCSRFToken [http-bio-8080-exec-10] in getCSRFToken 
2014-12-26 11:10:32,136 DEBUG c.t.t.a.c.LoginController.getCSRFToken [http-bio-8080-exec-10] CSRFToken set successfully. 
2014-12-26 11:12:40,500 INFO c.t.t.a.c.LoginController.getCSRFToken [http-bio-8080-exec-7] in getCSRFToken 
2014-12-26 11:12:40,501 DEBUG c.t.t.a.c.LoginController.getCSRFToken [http-bio-8080-exec-7] CSRFToken set successfully. 
2015-11-30 16:21:09,145 INFO c.t.t.s.a.i.AnalyticsServiceImpl.captureHit [http-bio-8080-exec-9] EnquiryDetails : {"createdTime":1448880669029,"modifiedTime":null,"active":true,"deleted":false,"deletedOn":-1,"guid":null,"uuid":null,"id":130771,"instanceId":130665,"pos":"","channel":"Web","flightNo":"TWBL2DL2","orig":"BLR","dest":"DEL","cabCls":"ECONOMY","logCls":"Y","noOfPaxs":1,"scheduleEntryId":130661,"travelDateTime":[2015,12,1,21,30],"enquiryDateTime":[2015,11,30,16,21,9,23000000]} 

Вы заметите, что последняя строка состоит из некоторых данных JSON Я пытаюсь настроить мой logstash, чтобы извлечь эти данные в формате JSON Ниже мой logstash конфигурационный файл:

input { 
    file { 
    path => "C:/Users/TESTER/Desktop/files/test1.log" 
    type => "test" 
     start_position => "beginning" 
    } 
} 


filter { 
    grok { 
    match => [ "message" , "timestamp : %{DATESTAMP:timestamp}", "severity: %{WORD:severity}", "clazz: %{JAVACLASS:clazz}", "selco: %{NOTSPACE:selco}", "testerField: (?<ENQDTLS>EnquiryDetails :)"] 

     } 
} 


output { 
    elasticsearch { 
     hosts => "localhost" 
     index => "test1" 
    } 
    stdout {} 
} 

Однако это мой логарифмический вывод:

C:\logstash-2.0.0\bin>logstash -f test1.conf 
io/console not supported; tty will not be manipulated 
Default settings used: Filter workers: 2 
Logstash startup completed 
2016-01-08T08:02:02.029Z TW 2014-12-24 09:41:29,383 INFO c.t.t.a.c.LoginController.getCSRFToken [http-bio-8080-exec-4] in getCSRFToken 
2016-01-08T08:02:02.029Z TW 2014-12-24 09:44:26,607 INFO c.t.t.a.c.LoginController.getCSRFToken [http-bio-8080-exec-8] in getCSRFToken 
2016-01-08T08:02:02.029Z TW 2014-12-24 09:44:26,609 DEBUG c.t.t.a.c.LoginController.getCSRFToken [http-bio-8080-exec-8] CSRFToken set successfully. 
2016-01-08T08:02:02.029Z TW 2014-12-26 09:55:28,399 INFO c.t.t.a.c.LoginController.getCSRFToken [http-bio-8080-exec-9] in getCSRFToken 
2016-01-08T08:02:02.029Z TW 2014-12-26 09:55:28,401 DEBUG c.t.t.a.c.LoginController.getCSRFToken [http-bio-8080-exec-9] CSRFToken set successfully. 
2016-01-08T08:02:02.029Z TW 2014-12-26 11:10:32,135 INFO c.t.t.a.c.LoginController.getCSRFToken [http-bio-8080-exec-10] in getCSRFToken 
2016-01-08T08:02:02.029Z TW 2014-12-26 11:10:32,136 DEBUG c.t.t.a.c.LoginController.getCSRFToken [http-bio-8080-exec-10] CSRFToken set successfully. 
2016-01-08T08:02:02.029Z TW 2014-12-24 09:41:29,383 DEBUG c.t.t.a.c.LoginController.getCSRFToken [http-bio-8080-exec-4] CSRFToken set successfully. 
2016-01-08T08:02:02.029Z TW 2014-12-26 11:12:40,500 INFO c.t.t.a.c.LoginController.getCSRFToken [http-bio-8080-exec-7] in getCSRFToken 
2016-01-08T08:02:02.029Z TW 2015-11-30 16:21:09,145 INFO c.t.t.s.a.i.AnalyticsServiceImpl.captureHit [http-bio-8080-exec-9] EnquiryDetails : {"createdTime":1448880669029,"modifiedTime":null,"active":true,"deleted":false,"deletedOn":-1,"guid":null,"uuid":null,"id":130771,"instanceId":130665,"pos":"","channel":"Web","flightNo":"TWBL2DL2","orig":"BLR","dest":"DEL","cabCls":"ECONOMY","logCls":"Y","noOfPaxs":1,"scheduleEntryId":130661,"travelDateTime":[2015,12,1,21,30],"enquiryDateTime":[2015,11,30,16,21,9,23000000]} 
2016-01-08T08:02:02.029Z TW 2014-12-26 11:12:40,501 DEBUG c.t.t.a.c.LoginController.getCSRFToken [http-bio-8080-exec-7] CSRFToken set successfully. 

Может кто-нибудь скажите, пожалуйста, что я делаю неправильно здесь. Спасибо

Answer 1

Я нашел решение моей проблемы.

input { 
    file { 
    path => "C:/Users/TESTER/Desktop/elk Files 8-1-2015/test1.log" 
     start_position => "beginning" 
    } 
} 


filter { 
    grok { 

    match => {"message" => "%{DATESTAMP:timestamp} %{WORD:severity} %{JAVACLASS:clazz} %{NOTSPACE:selco} (?<ENQDTLS>EnquiryDetails :) (?<JSONDATA>.*)"} 

    add_tag => [ "ENQDTLS"] 


} 

    if "ENQDTLS" not in [tags] {    
    drop { } 
    } 

    mutate { 
    remove_tag => ["ENQDTLS"] 
    } 

    json { 
     source => "JSONDATA" 
    } 

    mutate { 
    remove_field => ["timestamp"] 
    remove_field => ["clazz"] 
    remove_field => ["selco"] 
    remove_field => ["severity"] 
    remove_field => ["ENQDTLS"] 
    remove_field => ["JSONDATA"] 
    } 

} 


output { 
    elasticsearch { 
     hosts => "localhost" 
     index => "test3" 
    } 
    stdout { 
    codec => rubydebug 
    } 
} 

Так что Im делает здесь отфильтровывать любую строку, которая не содержит ключевое слово «EnquiryDetails» с помощью ГРОК, то я обработки данных в формате JSON в этой строке. Надеюсь, это поможет любому, у кого может быть такая же проблема. Также, поскольку я новичок в этом. хотел бы знать, является ли это хорошим подходом.

Answer 2

Вы не говорите, что вы испытываете, это «неправильно», но давайте предположим, что вас беспокоит нехватка полей в вашем выходе.

Во-первых, используйте rubydebug или json codec в выводе stdout {}. Он покажет вам более подробную информацию.

Во-вторых, похоже, что ваш grok {} все испорчен. grok {} принимает поле ввода и одно или несколько регулярных выражений для применения к входу. Вы даете ему вход («сообщение»), но это регулярное выражение:

"timestamp : %{DATESTAMP:timestamp}" 

не соответствует вводу, так как у вас нет строкового «метки времени:».

Вам нужно что-то подобное:

"%{DATESTAMP} %{WORD:severity}" (etc) 

Я хотел бы рекомендовать создание один Grok {} строфу, чтобы вытащить всю общую информацию от (все вплоть до]). Затем используйте другой, чтобы иметь дело с различными типами сообщений.