A look into metric collection overhead in logstash 5.4.x

## TL;DR

While most metrics in logstash-core are taken once per batch, because input plugins don't produce events in batches, the `WrappedWriteClient#<<` executes [`WrappedWriteClient#record_metrics`](https://github.com/elastic/logstash/blob/master/logstash-core/lib/logstash/instrument/wrapped_write_client.rb#L31) for every event, introducing almost a 40% performance drop in the tests below.

### Purpose

Study the impact of metric collection in event per second throughput in the latest logstash stable version

### Test method

I set out to compare the event throughput in a `generator -> elasticsearch` pipeline.
For this I used logstash 5.4 with the lastest elasticsearch output.
To remove the influence from network layer, I also modified the elasticsearch output to not perform the bulk request but always return a successful response.

### Test setup:

#### Logstash

Downloaded logstash 5.4.0 and updated the logstash-output-elasticsearch to 7.2.1

`bin/logstash-plugin install --version "7.2.1" logstash-output-elasticsearch`

##### Mocking network call in elasticsearch output

```patch
diff --git a/lib/logstash/outputs/elasticsearch/http_client.rb b/lib/logstash/outputs/elasticsearch/http_client.rb
index daca851..714c4b1 100644
--- a/lib/logstash/outputs/elasticsearch/http_client.rb
+++ b/lib/logstash/outputs/elasticsearch/http_client.rb
@@ -135,12 +135,14 @@ module LogStash; module Outputs; class ElasticSearch;
def bulk_send(body_stream)
params = http_compression ? {:headers => {"Content-Encoding" => "gzip"}} : {}
# Discard the URL
- _, response = @pool.post(@bulk_path, params, body_stream.string)
+ #_, response = @pool.post(@bulk_path, params, body_stream.string)
+ body_stream.string
if !body_stream.closed?
body_stream.truncate(0)
body_stream.seek(0)
end
- LogStash::Json.load(response.body)
+ #LogStash::Json.load(response.body)
+ {"took" => 30, "errors" => false, "items" => [] }
end

def get(path)
```
##### elasticsearch

Even though the _bulk requests are mocked, the rest aren't (template install, health checks). So elasticsearch must be running on 9200

#### Logstash Pipeline Configuration

```
input {
# 10 million messages
generator { message => '64.242.88.10 - - [07/Mar/2004:16:05:49 -0800] "GET /twiki/bin/edit/Main/Double_bounce_sender?topicparent=Main.ConfigurationVariables HTTP/1.1" 401 12846' count => 10000000 }
}
output {
# es output with bulk network call mocked out. see .patch files
elasticsearch {
hosts => "localhost"
index => "logstash-%{+YYYY.MM.dd.HH}"
manage_template => false
}
}
```

### Test executions

a) With metric collection enabled by setting `metric.collect: true` in the logstash.yml

```
/tmp/tests/logstash-5.4.0_es_7x % bin/logstash -f ../cfg -w 4 -b 125
Sending Logstash's logs to /tmp/tests/logstash-5.4.0_es_7x/logs which is now configured via log4j2.properties
[2017-06-02T09:18:16,748][INFO ][logstash.outputs.elasticsearch] Elasticsearch pool URLs updated {:changes=>{:removed=>[], :added=>[http://localhost:9200/]}}
[2017-06-02T09:18:16,753][INFO ][logstash.outputs.elasticsearch] Running health check to see if an Elasticsearch connection is working {:healthcheck_url=>http://localhost:9200/, :path=>"/"}
[2017-06-02T09:18:16,889][WARN ][logstash.outputs.elasticsearch] Restored connection to ES instance {:url=>#}
[2017-06-02T09:18:16,891][INFO ][logstash.outputs.elasticsearch] Using mapping template from {:path=>nil}
[2017-06-02T09:18:16,945][INFO ][logstash.outputs.elasticsearch] Attempting to install template {:manage_template=>{"template"=>"logstash-*", "version"=>50001, "settings"=>{"index.refresh_interval"=>"5s"}, "mappings"=>{"_default_"=>{"_all"=>{"enabled"=>true, "norms"=>false}, "dynamic_templates"=>[{"message_field"=>{"path_match"=>"message", "match_mapping_type"=>"string", "mapping"=>{"type"=>"text", "norms"=>false}}}, {"string_fields"=>{"match"=>"*", "match_mapping_type"=>"string", "mapping"=>{"type"=>"text", "norms"=>false, "fields"=>{"keyword"=>{"type"=>"keyword"}}}}}], "properties"=>{"@timestamp"=>{"type"=>"date", "include_in_all"=>false}, "@version"=>{"type"=>"keyword", "include_in_all"=>false}, "geoip"=>{"dynamic"=>true, "properties"=>{"ip"=>{"type"=>"ip"}, "location"=>{"type"=>"geo_point"}, "latitude"=>{"type"=>"half_float"}, "longitude"=>{"type"=>"half_float"}}}}}}}}
[2017-06-02T09:18:16,956][INFO ][logstash.outputs.elasticsearch] New Elasticsearch output {:class=>"LogStash::Outputs::ElasticSearch", :hosts=>[#]}
[2017-06-02T09:18:16,966][INFO ][logstash.pipeline ] Starting pipeline {"id"=>"main", "pipeline.workers"=>4, "pipeline.batch.size"=>125, "pipeline.batch.delay"=>5, "pipeline.max_inflight"=>500}
[2017-06-02T09:18:16,969][INFO ][logstash.pipeline ] Pipeline main started
[2017-06-02T09:18:17,046][INFO ][logstash.agent ] Successfully started Logstash API endpoint {:port=>9600}
[2017-06-02T09:24:04,712][WARN ][logstash.agent ] stopping pipeline {:id=>"main"}
bin/logstash -f ../cfg -w 4 -b 125 643.02s user 66.43s system 196% cpu 6:01.08 total
```

b) With metric collection disabled by setting `metric.collect: false` in the logstash.yml

```
/tmp/tests/logstash-5.4.0_es_7x % bin/logstash -f ../cfg -w 4 -b 125
Sending Logstash's logs to /tmp/tests/logstash-5.4.0_es_7x/logs which is now configured via log4j2.properties
[2017-06-02T09:01:50,573][INFO ][logstash.outputs.elasticsearch] Elasticsearch pool URLs updated {:changes=>{:removed=>[], :added=>[http://localhost:9200/]}}
[2017-06-02T09:01:50,579][INFO ][logstash.outputs.elasticsearch] Running health check to see if an Elasticsearch connection is working {:healthcheck_url=>http://localhost:9200/, :path=>"/"}
[2017-06-02T09:01:50,720][WARN ][logstash.outputs.elasticsearch] Restored connection to ES instance {:url=>#}
[2017-06-02T09:01:50,722][INFO ][logstash.outputs.elasticsearch] Using mapping template from {:path=>nil}
[2017-06-02T09:01:50,787][INFO ][logstash.outputs.elasticsearch] Attempting to install template {:manage_template=>{"template"=>"logstash-*", "version"=>50001, "settings"=>{"index.refresh_interval"=>"5s"}, "mappings"=>{"_default_"=>{"_all"=>{"enabled"=>true, "norms"=>false}, "dynamic_templates"=>[{"message_field"=>{"path_match"=>"message", "match_mapping_type"=>"string", "mapping"=>{"type"=>"text", "norms"=>false}}}, {"string_fields"=>{"match"=>"*", "match_mapping_type"=>"string", "mapping"=>{"type"=>"text", "norms"=>false, "fields"=>{"keyword"=>{"type"=>"keyword"}}}}}], "properties"=>{"@timestamp"=>{"type"=>"date", "include_in_all"=>false}, "@version"=>{"type"=>"keyword", "include_in_all"=>false}, "geoip"=>{"dynamic"=>true, "properties"=>{"ip"=>{"type"=>"ip"}, "location"=>{"type"=>"geo_point"}, "latitude"=>{"type"=>"half_float"}, "longitude"=>{"type"=>"half_float"}}}}}}}}
[2017-06-02T09:01:50,806][INFO ][logstash.outputs.elasticsearch] New Elasticsearch output {:class=>"LogStash::Outputs::ElasticSearch", :hosts=>[#]}
[2017-06-02T09:01:50,808][INFO ][logstash.pipeline ] Starting pipeline {"id"=>"main", "pipeline.workers"=>4, "pipeline.batch.size"=>125, "pipeline.batch.delay"=>5, "pipeline.max_inflight"=>500}
[2017-06-02T09:01:50,833][INFO ][logstash.pipeline ] Pipeline main started
[2017-06-02T09:01:50,934][INFO ][logstash.agent ] Successfully started Logstash API endpoint {:port=>9600}
[2017-06-02T09:04:50,222][WARN ][logstash.agent ] stopping pipeline {:id=>"main"}
bin/logstash -f ../cfg -w 4 -b 125 447.07s user 60.67s system 264% cpu 3:12.27 total
```

### Result summary

|logstash version|parameters|metric collection|time between pipeline start and end (seconds)| total runtime wallclock (seconds) | ~ eps |
|-----------------|------------|:---:|:---:|:---:|:---:|
|5.4.0 w/ es 7.2.1| -w 4 -b 125|true| 347.743 | 357.16 | 28757|
|5.4.0 w/ es 7.2.1| -w 4 -b 125|false| 179.389 | 192.27 | 55744|