Send json message only from a log file to elastic seach

Elastic Stack Logstash
1

Hi,

Below is the output of a log file, I will need to send json message only from a log file, and add a filter geoip for the visitorip

cat visitor_cluster01server02.log
[ECOM] [2017-10-11 14:50:06,810] [ECOM] [DEBUG] [VisitorLogger] [ ---------- {"visitorip":"xxx.xxx.xx.x","currenttime":"2017-10-11 14:50:06"} ---------- ]
[ECOM] [2017-10-11 14:52:49,708] [ECOM] [DEBUG] [VisitorLogger] [ ---------- {"visitorip":"xxx.xxx.xx.x","currenttime":"2017-10-11 14:52:49"} ---------- ]
[ECOM] [2017-10-11 15:01:45,117] [ECOM] [DEBUG] [VisitorLogger] [ ---------- {"visitorip":"xxx.xxx.xx.x","currenttime":"2017-10-11 15:01:45"} ---------- ]
[ECOM] [2017-10-11 15:04:38,897] [ECOM] [DEBUG] [VisitorLogger] [ ---------- {"visitorip":"xxx.xxx.xx.x","currenttime":"2017-10-11 15:04:38"} ---------- ]
[ECOM] [2017-10-11 15:06:48,389] [ECOM] [DEBUG] [VisitorLogger] [ ---------- {"visitorip":"xxx.xxx.xx.x","currenttime":"2017-10-11 15:06:48"} ---------- ]
[ECOM] [2017-10-11 15:11:55,948] [ECOM] [DEBUG] [VisitorLogger] [ ---------- {"visitorip":"xxx.xxx.xx.x","currenttime":"2017-10-11 15:11:55"} ---------- ]
[ECOM] [2017-10-11 15:12:08,399] [ECOM] [DEBUG] [VisitorLogger] [ ---------- {"visitorip":"xxx.xxx.xx.x","currenttime":"2017-10-11 15:12:08"} ---------- ]
[ECOM] [2017-10-11 15:17:10,574] [ECOM] [DEBUG] [VisitorLogger] [ ---------- {"visitorip":"xxx.xxx.xx.x","currenttime":"2017-10-11 15:17:10"} ---------- ]
[ECOM] [2017-10-11 15:17:44,688] [ECOM] [DEBUG] [VisitorLogger] [ ---------- {"visitorip":"xxx.xxx.xx.x","currenttime":"2017-10-11 15:17:44"} ---------- ]
[ECOM] [2017-10-11 15:21:28,905] [ECOM] [DEBUG] [VisitorLogger] [ ---------- {"visitorip":"xxx.xxx.xx.x","currenttime":"2017-10-11 15:21:28"} ---------- ]
[ECOM] [2017-10-11 15:27:45,364] [ECOM] [DEBUG] [VisitorLogger] [ ---------- {"visitorip":"xxx.xxx.xx.x","currenttime":"2017-10-11 15:27:45"} ---------- ]
[ECOM] [2017-10-11 15:31:45,866] [ECOM] [DEBUG] [VisitorLogger] [ ---------- {"visitorip":"xxx.xxx.xx.x","currenttime":"2017-10-11 15:31:45"} ---------- ]
[ECOM] [2017-10-11 15:37:05,951] [ECOM] [DEBUG] [VisitorLogger] [ ---------- {"visitorip":"xxx.xxx.xx.x","currenttime":"2017-10-11 15:37:05"} ---------- ]
[ECOM] [2017-10-11 15:37:26,468] [ECOM] [DEBUG] [VisitorLogger] [ ---------- {"visitorip":"xxx.xxx.xx.x","currenttime":"2017-10-11 15:37:26"} ---------- ]

Below is another output of log file, I will need to send json message only to the elastic search, adding geoip filter to the visitorip.

cat orderinfo_cluster01server02.log
ECOM] [2017-10-12 11:02:22,694] [ECOM] [INFO] [CHECKOUT_LOGGER] [Order: 288835 - has been submitted to OMS.]
[ECOM] [2017-10-12 11:02:22,714] [ECOM] [DEBUG] [CHECKOUT_LOGGER] [the order service endpoint URL :::http://stage.enterprise.com/12c-order]
[ECOM] [2017-10-12 11:02:22,792] [ECOM] [DEBUG] [DeveloperDebugLogger] [ ---------- ServiceRegistryUtil.getEndpoint() --->>>> DataPacketService_V1 :---
[ECOM] [2017-10-12 11:02:22,794] [ECOM] [DEBUG] [DATA_PACKET_LOGGER] [{"request":{"orderNumber":675689359,"transactionDateTimeIso":"2017-10-12 11:02:22:000002","businessChannel":"INTERNET","trafficChannel":"INTERNET","totalOrderAmt":28.85,"merchAmt":0.00,"taxAmt":1.89,"shippingAmt":6.99,"paymentType":"Credit Card","promoNumber":null,"numberItemsInCart":1,"shippingZip":"68137","clientIP":"XXx.xx.xx.x","orderPacketDetail":[{"lineNumber":0,"sku":"136854","skuPrice":19.97}]}}]
[ECOM] [2017-10-12 11:02:28,012] [ECOM] [INFO] [CHECKOUT_LOGGER] [createnewacct@otc.net ProcessOrder:SubmitSuccess: Time Taken to place order to OMS in miliseconds.5318:OrderNumber:675689359]
[ECOM] [2017-10-12 11:02:28,012] [ECOM] [ERROR] [com.common.repository.util.EnvProperty] [PerformanceLogger is missing for CHECKOUT]
[ECOM] [2017-10-12 11:02:28,178] [ECOM] [INFO] [CHECKOUT_LOGGER] [createnewacct@otc.net ProcessOrder:ProcessCometResponse: Time Taken to process order in miliseconds.166:OrderNumber:675689359]
[ECOM] [2017-10-12 11:13:23,584] [ECOM] [DEBUG] [VisitorLogger] [ ---------- {"visitorip":"XXx.xx.xx.x","currenttime":"2017-10-12 11:13:23"} ---------- ]
[ECOM] [2017-10-12 11:20:29,495] [ECOM] [DEBUG] [VisitorLogger] [ ---------- {"visitorip":"XXx.xx.xx.x","currenttime":"2017-10-12 11:20:29"} ---------- ]
[ECOM] [2017-10-12 11:25:07,197] [ECOM] [DEBUG] [VisitorLogger] [ ---------- {"visitorip":"XXx.xx.xx.x","currenttime":"2017-10-12 11:25:07"} ---------- ]
[ECOM] [2017-10-12 11:25:12,178] [ECOM] [DEBUG] [DeveloperDebugLogger] [ ---------- ServiceRegistryUtil.getEndpoint() --->>>> CustomerService_V1 :--- http://qa3.enterprise.com/customer ---------- ]
[ECOM] [2017-10-12 11:25:12,226] [ECOM] [DEBUG] [DeveloperDebugLogger] [ ---------- ServiceRegistryUtil.getEndpoint() --->>>> EmailSubscriptionService_V1 :--- http://qa3.enterprise.com/emailsubscription ---------- ]

The above log files are on the same server, if possible can this done with the single logstash conf file?

2

Sure. You'll want to use a grok filter to extract timestamp and other interesting fields as well as the JSON payload, and then you run a json filter on the field with the extracted JSON string. A single grok filter can contain multiple expressions that'll get tried in order so parsing logs that look slightly differently shouldn't be a problem.

3

Hello Magnusbaeck,
I am new the logstash, tried various patterns,the log file pattern got changed with below, I need only to extract the json from the log and send it to elastic search, can you help me with this.

[Name] [2017-10-13 14:51:57,091] [Name] [DEBUG] [CustomLogger] [{"visitorip":"141.8.144.80","currenttime":"2017-10-13 14:51:57"}]

image
image.png1057×561 15 KB

[Name] [2017-10-13 11:15:36,011] [Name] [DEBUG] [com.enterprise.business.client.InfoClient] [{"request":{"orderNumber":122389405,"transactionDateTimeIso":"2017-10-13 11:15:36:000015","businessChannel":"INTERNET","trafficChannel":"INTERNET","totalOrderAmt":62.03,"merchAmt":0.00,"taxAmt":4.06,"shippingAmt":9.99,"paymentType":"Credit Card","promoNumber":null,"numberItemsInCart":2,"shippingZip":"36608","clientIP":"186.16.21.20","orderPacketDetail":[{"lineNumber":0,"sku":"3/12808","skuPrice":22.99},{"lineNumber":1,"sku":"4946","skuPrice":24.99}]}}]

image
image.png1023×575 31.2 KB

4

tried with below pattern for first log.

%{WORD:name} %{TIMESTAMP_ISO8601:LogDate} %{WORD:name} %{LOGLEVEL:loglevel} %{WORD:visitor} %{GREEDYDATA:Line}

5

The grok constructor web site can help you construct a grok expression for your log.

6

below is my logstash config, I am able to get the json data, but not able to parse the geoip from the json, as it is nested json, can you help me with that, I need to remove nested parent filed request from json to get the geo ip from client ip

input {
beats {
port => "5043"
}
}
filter {
grok {
match => { "message" => "[%{WORD:module}] [%{TIMESTAMP_ISO8601:LogDate}] [%{WORD:ecom1}] [%{LOGLEVEL:loglevel}] [%{JAVACLASS:logclass}] [%{GREEDYDATA:Line}]"}
}
json {
source => "Line"
}
geoip {
source => "clientIP"
}
if "*failure" in [tags] {
drop { }
}
}
output {
stdout { codec => rubydebug }
}

7

I got it working after adding

geoip {
source => "[request][clientIP]"
}

8

This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.