Logstash performance degrades as I add in a filter and output

[k1ng87]

I'm noticing something very odd in performance in logstash. If I have my config file set like this:

input { 
        kafka { 
                topics => ["kafka-jmx"]
                bootstrap_servers => "kafka1.com:9092"
                consumer_threads => 1
        }
 
}
output {
                stdout {}
}

my consumption is about 20k messages/sec from kafka. I can see that because I started logstash with an RMI listener so I can see whats going on in the JVM via jconsole.

As soon as I add a filter in like this:

filter {
        json {  
                source => "message"
        }
        grok {  
                patterns_dir => "/home/ec2-user/logstash-5.2.0/bin/patterns/"
                match => {"metric_path" => [
                                "%{DATA:kafka_host}\.%{DATA:kafka_metric_group}:type=%{DATA:kafka_metric_type},name=%{WORD:kafka_metric_name},topic=%{KTOPIC:kafka_topic},partition=%{KPARTITION:topic_partition}\.%{GREEDYDATA:attr_type}",
                                "%{DATA:kafka_host}\.%{DATA:kafka_metric_group}:type=%{DATA:kafka_metric_type},name=%{WORD:kafka_metric_name},topic=%{KTOPIC:kafka_topic}\.%{GREEDYDATA:attr_type}",
                                "%{DATA:kafka_host}\.%{DATA:kafka_metric_group}:type=%{DATA:kafka_metric_type},name=%{GREEDYDATA:kafka_metric_name}\.%{GREEDYDATA:attr_type}"
                                ]
                         }
        }
               ruby {   
                        code => "event.set('time', event.get('@timestamp').to_f * 1000 )"
                }
                mutate {
                        remove_field => ["message"]
                        convert => {"time" => "integer"
                                    "metric_value_number" => "integer"
                        }
                }
}

It goes from 20k/sec to about 1,500/sec

And then when I add the output in like this:

output {
        influxdb {
                host => "10.204.95.88"
                db => "monitoring"
                measurement => "BrokerMetrics"
                retention_policy => "one_week"
                allow_time_override => "true"
                exclude_fields => ["@timestamp", "@version", "path"]
                data_points => {
                                "time" => "%{time}"
                                "cluster_field" => "%{cluster}"
                                "kafka_host_field" => "%{kafka_host}"
                                "kafka_metric_group_field" => "%{kafka_metric_group}"
                                "kafka_metric_type_field" => "%{kafka_metric_type}"
                                "kafka_metric_name_field" => "%{kafka_metric_name}"
                                "kafka_topic_field" => "%{kafka_topic}"
                                "attr_type_field" => "%{attr_type}"
                                "cluster" => "%{[cluster]}"
                                "kafka_host" => "%{[kafka_host]}"
                                "kafka_metric_group" => "%{[kafka_metric_group]}"
                                "kafka_metric_type" => "%{[kafka_metric_type]}"
                                "kafka_metric_name" => "%{[kafka_metric_name]}"
                                "kafka_topic" => "%{[kafka_topic]}"
                                "attr_type" => "%{[attr_type]}"
                                "metric_value_number" => "%{metric_value_number}"
                                "metric_value_string" => "%{metric_value_string}"
                                "topic_partition_field" => "%{topic_partition}"
                                "topic_partition" => "%{[topic_partition]}"
                        }
                coerce_values => {"metric_value_number" => "integer"}
                send_as_tags => [ "kafka_host", "kafka_metric_group","cluster", "kafka_metric_type", "kafka_metric_name", "attr_type", "kafka_topic", "topic_partition" ]
                }
       }

consumption drops down from 1,500/sec to about 300/sec. So all in all, my rate drops from 20,000/sec to 300/sec.

I don't have any settings set in my logstash.yml file and I set the heap_size to 2g (and the jvm tells me I have plenty of heap space). I'm also only using about 60% cpu usage as well...

Why is this happening? I've also tried with -w 2 and all the way up to 4 when starting logstash but that seemed to have no affect....