Geoip in logstash returns coordinates correctly for simple grok filter but not from more complex filter

Elastic Stack Logstash
#1

Hi,

I am trying to extract an IP address from a syslog message in log stash. When I run log stash in debugging mode to stdout using only this simple filter it maps to geoip correctly and the geoip plugin processes and outputs additional calculated fields:

filter {

grok {
match => { "message" => "%{IPV4:ip}"}
}
geoip {
source => "ip"
}
}

Input I am testing with:

Mar 12 04:05:30 ip-172-31-20-78 2018-03-12 message repeated 991 times: [04:05:30.109066 [!] Artillery has detected an attack from IP address: 114.115.146.121 for a connection on a honeypot port: 1433]
Mar 12 04:05:35 ip-172-31-20-78 2018-03-12 message repeated 1000 times: [04:05:35.252924 [!] Artillery has detected an attack from IP address: 114.115.146.121 for a connection on a honeypot port: 1433]
Mar 12 04:05:35 ip-172-31-20-78 2018-03-12 message repeated 1001 times: [04:05:35.692720 [!] Artillery has detected an attack from IP address: 114.115.146.121 for a connection on a honeypot port: 1433]
Mar 12 04:05:36 ip-172-31-20-78 2018-03-12 message repeated 1002 times: [04:05:36.270937 [!] Artillery has detected an attack from IP address: 114.115.146.121 for a connection on a honeypot port: 1433]

However when I adapt this pattern to my syslog logstash filter it is not returning the field or any of the latitude and longitude calculations like it does when using the simple filter. Here is the filter I am referencing:
filter {
if [type] == "syslog" {
grok {
match => { "message" => "%{SYSLOGTIMESTAMP:syslog_timestamp} %{SYSLOGHOST:syslog_hostname} %{DATA:syslog_program}(?:[%{POSINT:syslog_pid}])?: %{IPV4:ip} %{GREEDYDATA:syslog_message} " }
add_field => [ "received_at", "%{@timestamp}" ]
add_field => [ "received_from", "%{host}" ]
}
geoip {source => "ip" }

syslog_pri { }

date {match => [ "syslog_timestamp", "MMM d HH:mm:ss", "MMM dd HH:mm:ss" ]}

}
}

Now when I test the grok filter using the grok debugger it is indeed parsing out the ip field from the message as desired. Here is the output from the grok debugger:

{
"syslog_timestamp": [
[
"Mar 12 04:05:30"
]
],
"MONTH": [
[
"Mar"
]
],
"MONTHDAY": [
[
"12"
]
],
"TIME": [
[
"04:05:30"
]
],
"HOUR": [
[
"04"
]
],
"MINUTE": [
[
"05"
]
],
"SECOND": [
[
"30"
]
],
"syslog_hostname": [
[
"ip-172-31-20-78"
]
],
"IPORHOST": [
[
"ip-172-31-20-78"
]
],
"HOSTNAME": [
[
"ip-172-31-20-78"
]
],
"IP": [
[
null
]
],
"IPV6": [
[
null
]
],
"IPV4": [
[
null
]
],
"syslog_program": [
[
"2018-03-12 message repeated 991 times: [04:05:30.109066 [!] Artillery has detected an attack from IP address"
]
],
"syslog_pid": [
[
null
]
],
"ip": [
[
"114.115.146.121"
]
],
"syslog_message": [
[
"for a connection on a honeypot port:"
]
]
}

I am trying to determine why the parsing and assignment to the geoip field doesn't work in the updated logstash.conf file considering the more simplistic version is working correctly.

Thanks,

Chad

#2

Don't use DATA to capture syslog_program. As you can see it's clearly matching too much. Have you tried SYSLOGPROG?

Regarding the IP address, do you want to pick up anything that looks like an IP address in the message?

#3

First, you really should use the </> tag to format you copy / paste text. Makes it much more readable. Secondly, in you "simplified" version you do not test the type. My guess would be that the logline you receiving is not of type syslog.

What happens if you remove the if statement in you more "complex" version?

#4

Hi Magnus,

Thanks for your response. Yes, I want to pick up anything that looks like an ip address in the message. I will try to test using SYSLOGPROG today.

#5

I suggest you use the IP pattern in a separate grok filter that looks at the syslog_message field.

#6

Would this be via an additional condition within the filter or a separate filter file (I didn't think multiple filter files were supported).

As a side note I have been testing with multiple pipelines as of yesterday. I have one pipeline for the 'complex' filter which is accepting filebeat input and outputting to elasticsearch. (The syslog filter) . I have created an additional pipeline with the simple geoip filter that calculates coordinates in debugging mode and it is set to use stdin as input and it is also set to output to elasticsearch. I've noticed that in this scenario (the stdin geoip filter) nothing is surfacing in kibana. I don't know how to determine if it is making it's way into elasticsearch into kibana and where the disconnect is.

I

#7

Would this be via an additional condition within the filter or a separate filter file (I didn't think multiple filter files were supported).

You need two separate grok filters. They could be in the same configuration file or in different files. If you choose to put them in separate files just keep in mind that configuration files are read in alphabetical order and you obviously want the grok filter that creates the syslog_message field to run before the grok filter that attempts to parse the same field.

#8

Hi Magnus. I just solved the issue. I ended up creating a second pipeline in the pipeline.yml with an input of stdinput to test. The coordinates started showing up in Kibana after creating a new index. But the next issue was that the geoip.location field was missing. I found an elastic search template with the correct mappings and configured the output of this second pipeline to point to this template path and marked it as 'managed'. After restarting logstash, piping in more data and refreshing the indexes in kibana the geoip.location field is now surfacing in the kibana dashboard with a type of geo_point. Subsequently I created a tile map visualization and it showed on the tile map successfully. Thanks for your help and hopefully this information is helpful to someone else. - chad

#9

And my next step will be to substitute the stdinput of the second pipeline with a filbert input using a different port. I am going to configure the filebeat agent to output to two different logstash outputs using these different ports. Hopefully that will work as anticipated. I believe I've read that this is possible in the filbeats config.

#10

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