Logstash spikes to ~800% when adding one more GROK expression?

Elastic Stack Logstash
1

Hi All,

I'm suffering a strange issue when trying to process IDS events with Logstash.
My events are essentially , delimited as follows;

2016-03-11T22:18:48.559342+00:00 1457648327 2016-03-11T22:18:48.538761+00:00 1457648327 ukloninfosecsfs01 SFIDSXYSC: Protocol: TCP, SrcIP: 10.10.10.10, DstIP: 10.10.10.11, SrcPort: 88, DstPort: 21586, TCPFlags: 0x0, IngressInterface: s3p4, IngressZone: MASKEDvalue, DE: Primary Detection Engine (352b0534-b660-11e4-b7f2-efa234d91272), Policy: XY BO Remote, ConnectType: End, AccessControlRuleName: BO General, AccessControlRuleAction: Allow, UserName: No Authentication Required, InitiatorPackets: 4, ResponderPackets: 0, InitiatorBytes: 256, ResponderBytes: 0, DNSResponseType: No Error, Sinkhole: Unknown, URLCategory: Unknown, URLReputation: Risk unknown

All IDS events contain the following prefix format.

2016-03-11T22:18:48.559342+00:00 1457648327 2016-03-11T22:18:48.538761+00:00 1457648327 ukloninfosecsfs01 SFIDSXYSC: Protocol: TCP, SrcIP: 10.10.10.10, DstIP: 10.10.10.11, SrcPort: 88, DstPort: 21586, TCPFlags: 0x0, IngressInterface: s3p4, IngressZone: MASKEDvalue, DE: Primary Detection Engine (352b0534-b660-11e4-b7f2-efa234d91272), Policy: XY BO Remote, ConnectType: End, AccessControlRuleName: BO General, AccessControlRuleAction: Allow, UserName: No Authentication Required,

And only the section after this tends to change with a small variety of additional fields being inserted dependant on protocol/client etc.

I have written my GROK as follows;

^\S+\s\S+\s%{TIMESTAMP_ISO8601:time_3}\s\S+\s(?:%{SYSLOGFACILITY} )?%{SYSLOGHOST:logsource} %{SYSLOGPROG}:\s+\w+:\s+%{WORD:protocol},\s\w+:\s+%{IP:src_ip},\s+\w+:\s+%{IP:dst_ip},\s+\w+:\s+%{INT:src_port},\s+\w+:\s%{INT:dst_port},\s+\w+:\s%{BASE16NUM:tcpflags},\s+\w+:\s%{WORD:ingress_interface},\s+\w+:\s+%{WORD:ingress_zone},\s+\w+:\s+(?[^,]+),\s+\w+:\s%{DATA:policy},\s+\w+:\s%{DATA:connect_type},\s+\w+:\s%{DATA:access_control_rule_name},\s+\w+:\s%{DATA:access_control_rule_action},\sUserName:\s%{DATA:username},\s+InitiatorPackets:\s%{INT:initiator_packets},\s+ResponderPackets:\s%{INT:responder_packets},\s+InitiatorBytes:\s%{INT:initiator_bytes},\s+ResponderBytes:\s%{INT:responder_bytes},

Note I have defined some of the above as patterns to reduce the complexity within my .conf files. The patterns have been removed here.
The above works well and processes a significant number of EPS.

The above only matches to the point ResponderBytes: 0,

If I append _s+DNSResponseType:\s%{DATA:DNSRespType},_ the CPU spikes to 800% and nothing is processed. This is one example. There is a number of other variations I’ve tried however essentially I cannot progress by adding another matched field.

I can’t see why one more field would seemly take down logstash.

Data is being feed into Elastic.

I’ve trying tuning work threads, logstash heap etc.
Your help would be most appreciated.

2

It looks to me like everything after this is a sequence of key-value pairs. You might be able to use grok to parse the fields before this and capture this key-value list in a single field and apply the kv filter instead.

3

Thanks for the guidance,
To be honest KV isn't a filter I've used before.

I've been experimenting following your post, but I'm struggling to manage whitespace in my 'Values'. KV will spot each of my Keys with ease but trims at the first whitespace. Is there a trick I'm missing?

Examples.
"URLReputation" => "Risk" , It should be "URLReputation" => "Risk unknown"
"DE" => "Primary", it should be "DE" => "Primary Detection Engine (352b0534-b660-11e4-b7f2-efa234d91272)"

Thanks

4

Having just posted that,
I swapped the KV separator to = and then used

filter {
grok {
match => [
"message" , "^\S+\s\S+\s%{TIMESTAMP_ISO8601:time_3}\s\S+\s(?:%{SYSLOGFACILITY} )?%{SYSLOGHOST:logsource} %{SYSLOGPROG}:\s+%{GREEDYDATA:pending}"
]
}

	kv { 
		source => "pending"
		field_split => ","

		}

}

This solved the whitespace issue.
I need to combine that with the correct ':' delimiter now.

5

Sorted,
kv {
source => "pending"
field_split => ","
value_split => ":"
}

Thanks for the help!