Record the execution time for each output in the pipeline

Record the wall clock time for each output a new `duration_in_millis`
key will now be available for each output in the api located at http://localhost:9600/_node/stats

This commit also change some expectations in the output_delegator_spec
that were not working as intended with the `have_received` matcher.

Fixes #6458

logstash-core/lib/logstash/output_delegator.rb

@@ -39,7 +39,9 @@ module LogStash class OutputDelegator
 
   def multi_receive(events)
     @metric_events.increment(:in, events.length)
+    clock = @metric_events.time(:duration_in_millis)
     @strategy.multi_receive(events)
+    clock.stop
     @metric_events.increment(:out, events.length)
   end
 

logstash-core/spec/logstash/output_delegator_spec.rb

@@ -45,16 +45,24 @@ describe LogStash::OutputDelegator do
     context "after having received a batch of events" do
       before do
         subject.register
-        subject.multi_receive(events)
       end
 
       it "should pass the events through" do
-        expect(out_inst).to have_received(:multi_receive).with(events)
+        expect(out_inst).to receive(:multi_receive).with(events)
+        subject.multi_receive(events)
       end
 
       it "should increment the number of events received" do
-        expect(subject.metric_events).to have_received(:increment).with(:in, events.length)
-        expect(subject.metric_events).to have_received(:increment).with(:out, events.length)
+        expect(subject.metric_events).to receive(:increment).with(:in, events.length)
+        expect(subject.metric_events).to receive(:increment).with(:out, events.length)
+        subject.multi_receive(events)
+      end
+
+      it "should record the `duration_in_millis`" do
+        clock = spy("clock")
+        expect(subject.metric_events).to receive(:time).with(:duration_in_millis).and_return(clock)
+        expect(clock).to receive(:stop)
+        subject.multi_receive(events)
       end
     end
 

logstash-core/spec/logstash/pipeline_spec.rb

@@ -756,7 +756,10 @@ describe LogStash::Pipeline do
 
       it "populates the output metrics" do
         plugin_name = dummy_output_id.to_sym
+
+        expect(collected_metric[:stats][:pipelines][:main][:plugins][:outputs][plugin_name][:events][:in].value).to eq(number_of_events)
         expect(collected_metric[:stats][:pipelines][:main][:plugins][:outputs][plugin_name][:events][:out].value).to eq(number_of_events)
+        expect(collected_metric[:stats][:pipelines][:main][:plugins][:outputs][plugin_name][:events][:duration_in_millis].value).not_to be_nil
       end
 
       it "populates the name of the output plugin" do