Hi,
I am quite new to Logstash , i was able to successfully push the messages from Logstash over to ElasticSearch, however the @timestamp is different from the one specified in logs. I want to set the log time to @timestamp . I tried parsing through previous topics as well, but couldn't figure out the same.
I am reading the logs from Kafka topic (in JSON Format) and sending over to ElasticSearch.
My query is to :- Fetch the timeInMilliseconds in logs , convert it in date format (UTC timezone) and set the same as @timestamp.
Any pointers to the same would be really helpful, Thanks a lot...!!!
Json logs on kafka
++++++++++++++++++++++++++
{"timeInMilliseconds":1503611308777,"level":"WARN","logger":"com.abcd.","thread":"http-nio-8080-exec-1","message":"The endpoint returned 401. The token will be regenerated and request executed once again","config":{"container":{"id":"2fb84e725af5"},"java":{"os":"Linux 4.4.57-4.4.1-amd64-089fa03c2695b87c unknown, architecture: amd64-64","vm":"Java HotSpot(TM) 64-Bit Server VM (build 25.121-b13, mixed mode)","runtime":"Java(TM) SE Runtime Environment (build 1.8.0_121-b13) from Oracle Corporation"},"host":{"name":"xxx.xxx.xxx"}},"contextMap":{"X-B3-TraceId":"dde262dd1f6cd3cd","X-Span-Export":"false","X-B3-SpanId":"dde262dd1f6cd3cd","session":"93c1847f2b76446374835f88aa983c6af7e1316"},"endOfBatch":false}
DEBUG logs
++++++++++++++++++++++++++
filter received {:event=>{"timeInMilliseconds"=>1504111665465, "level"=>"WARN", "logger"=>"com.abcd", "thread"=>"pool-5-thread-1", "message"=>"OutputRequest failed: [ex=, body={"code":"invalid_token","title":"Token is revoked","incidentId":"b25c3a01-7a9b-4650-93bf-d3447c590054","requestId":"5e0f122c-9d88-4a88-9915-0a4e42320d7a","session":"2daae1c7e4fb9e6ce5b42112f26e70ce98c869","solution":"request_token"}]", "config"=>{"container"=>{"id"=>"d7b1d7d630a8"}, "java"=>{"os"=>"Linux 4.4.57-4.4.1-amd64-089fa03c2695b87c unknown, architecture: amd64-64", "vm"=>"Java HotSpot(TM) 64-Bit Server VM (build 25.121-b13, mixed mode)", "runtime"=>"Java(TM) SE Runtime Environment (build 1.8.0_121-b13) from Oracle Corporation"}, "host"=>{"name"=>"xxx.xxx.xxx"}}, "contextMap"=>{"outSpanId"=>"c99db8d0-69cd-4faa-8132-20fa220bdfd1", "outUrl"=>"http://abcd.com", "component"=>"[IDS-SDK-RS]", "outMethod"=>"GET"}, "endOfBatch"=>false, "@version"=>"1", "@timestamp"=>"2017-08-30T16:48:26.005Z"}, :level=>:debug, :file=>"(eval)", :line=>"41", :method=>"filter_func"}
Date filter: received event {:type=>nil, :level=>:debug, :file=>"logstash/filters/date.rb", :line=>"311", :method=>"filter"}
Date filter looking for field {:type=>nil, :field=>"timeInMilliseconds", :level=>:debug, :file=>"logstash/filters/date.rb", :line=>"314", :method=>"filter"}
Date parsing done {:value=>1504111665465, :level=>:debug, :file=>"logstash/filters/date.rb", :line=>"348", :method=>"filter"}
filter received {:event=>{"timeInMilliseconds"=>1504111665465, "level"=>"WARN", "logger"=>"com.abcd", "thread"=>"pool-5-thread-1", "message"=>"The endpoint returned 401. The token will be regenerated and request executed once again", "config"=>{"container"=>{"id"=>"d7b1d7d630a8"}, "java"=>{"os"=>"Linux 4.4.57-4.4.1-amd64-089fa03c2695b87c unknown, architecture: amd64-64", "vm"=>"Java HotSpot(TM) 64-Bit Server VM (build 25.121-b13, mixed mode)", "runtime"=>"Java(TM) SE Runtime Environment (build 1.8.0_121-b13) from Oracle Corporation"}, "host"=>{"name"=>"xxx.xxx.xxx"}}, "endOfBatch"=>false, "@version"=>"1", "@timestamp"=>"2017-08-30T16:48:26.006Z"}, :level=>:debug, :file=>"(eval)", :line=>"41", :method=>"filter_func"}
Date filter: received event {:type=>nil, :level=>:debug, :file=>"logstash/filters/date.rb", :line=>"311", :method=>"filter"}
Date filter looking for field {:type=>nil, :field=>"timeInMilliseconds", :level=>:debug, :file=>"logstash/filters/date.rb", :line=>"314", :method=>"filter"}
Date parsing done {:value=>1504111665465, :level=>:debug, :file=>"logstash/filters/date.rb", :line=>"348", :method=>"filter"}
{
"timeInMilliseconds" => 1504111665465,
"level" => "WARN",
"logger" => "com.abcd",
"thread" => "pool-5-thread-1",
"message" => "OutputRequest failed: [ex=, body={"code":"invalid_token","title":"Token is revoked","incidentId":"b25c3a01-7a9b-4650-93bf-d3447c590054","requestId":"5e0f122c-9d88-4a88-9915-0a4e42320d7a","session":"2daae1c7e4fb9e6ce5b42112f26e70ce98c869","solution":"request_token"}]",
"config" => {
"container" => {
"id" => "d7b1d7d630a8"
},
"java" => {
"os" => "Linux 4.4.57-4.4.1-amd64-089fa03c2695b87c unknown, architecture: amd64-64",
"vm" => "Java HotSpot(TM) 64-Bit Server VM (build 25.121-b13, mixed mode)",
"runtime" => "Java(TM) SE Runtime Environment (build 1.8.0_121-b13) from Oracle Corporation"
},
"host" => {
"name" => "xxx.xxx.xxx"
}
},
"contextMap" => {
"outSpanId" => "c99db8d0-69cd-4faa-8132-20fa220bdfd1",
"outUrl" => "http://abcd.com",
"outTraceId" => "5e0f122c-9d88-4a88-9915-0a4e42320d7a",
"component" => "[IDS-SDK-RS]",
"outMethod" => "GET"
},
"endOfBatch" => false,
"@version" => "1",
"@timestamp" => "2017-08-30T16:47:45.465Z"
}
Logstash conf
++++++++++++++++++++++++++
input {
kafka {
zk_connect => "11.1.11.11"
topic_id => "test"
codec => "json"
reset_beginning => true
}
}
filter {
date {
timezone => "UTC"
match => ["timeInMilliseconds", "UNIX_MS"]
target => "@timestamp"
}
}
output {
stdout { codec => rubydebug }
}