From b1bac4eeb805a7326429c1023130eb6fd9e0abd7 Mon Sep 17 00:00:00 2001
From: Aleksandr Chermenin <aleksandr_chermenin@epam.com>
Date: Tue, 31 Jul 2018 11:48:12 +0500
Subject: [PATCH] Added option 'keep_start_event'

---
 docs/index.asciidoc             | 12 +++++++++++
 lib/logstash/filters/elapsed.rb |  8 +++++++-
 spec/filters/elapsed_spec.rb    | 36 +++++++++++++++++++++++++--------
 3 files changed, 47 insertions(+), 9 deletions(-)

diff --git a/docs/index.asciidoc b/docs/index.asciidoc
index ffd9403..7760d91 100644
--- a/docs/index.asciidoc
+++ b/docs/index.asciidoc
@@ -106,6 +106,7 @@ This plugin supports the following configuration options plus the <<plugins-{typ
 | <<plugins-{type}s-{plugin}-start_tag>> |<<string,string>>|Yes
 | <<plugins-{type}s-{plugin}-timeout>> |<<number,number>>|No
 | <<plugins-{type}s-{plugin}-unique_id_field>> |<<string,string>>|Yes
+| <<plugins-{type}s-{plugin}-keep_start_event>> |<<string,string>>|No
 |=======================================================================
 
 Also see <<plugins-{type}s-{plugin}-common-options>> for a list of options supported by all
@@ -162,6 +163,17 @@ The name of the field containing the task ID.
 This value must uniquely identify the task in the system, otherwise
 it's impossible to match the couple of events.
 
+[id="plugins-{type}s-{plugin}-keep_start_event"]
+===== `keep_start_event`
+
+  * Value type is <<string,string>>
+  * Default value is `first`
+
+This property manages what to do when several events matched as a start one
+were received before the end event for the specified ID. There are two
+supported values: `first` or `last`. If it's set to `first` (default value),
+the first event matched as a start will be used; if it's set to `last`,
+the last one will be used.
 
 
 [id="plugins-{type}s-{plugin}-common-options"]
diff --git a/lib/logstash/filters/elapsed.rb b/lib/logstash/filters/elapsed.rb
index b270b16..7e9456c 100644
--- a/lib/logstash/filters/elapsed.rb
+++ b/lib/logstash/filters/elapsed.rb
@@ -109,6 +109,12 @@ class LogStash::Filters::Elapsed < LogStash::Filters::Base
   # to the "end event"; if it's set to `true` a new "match event" is created.
   config :new_event_on_match, :validate => :boolean, :required => false, :default => false
 
+  # This property manage what to do when several "start events" were received
+  # before an "end event" for a concrete ID. If it's set to `first` (default
+  # value), the first "start event" will be used; if it's set to `last`,
+  # the last "start event" will be used.
+  config :keep_start_event, :validate => ['first', 'last'], :required => false, :default => 'first'
+
   # This filter must have its flush function called periodically to be able to purge
   # expired stored start events.
   config :periodic_flush, :validate => :boolean, :default => true
@@ -139,7 +145,7 @@ def filter(event)
       @logger.debug("Elapsed, 'start event' received", start_tag: @start_tag, unique_id_field: @unique_id_field)
 
       @mutex.synchronize do
-        unless(@start_events.has_key?(unique_id))
+        unless(@keep_start_event == 'first' && @start_events.has_key?(unique_id))
           @start_events[unique_id] = LogStash::Filters::Elapsed::Element.new(event)
         end
       end
diff --git a/spec/filters/elapsed_spec.rb b/spec/filters/elapsed_spec.rb
index 1f2ac9f..c1be2b2 100644
--- a/spec/filters/elapsed_spec.rb
+++ b/spec/filters/elapsed_spec.rb
@@ -74,16 +74,36 @@ def setup_filter(config = {})
     end
 
     describe "receiving two 'start events' for the same id field" do
-      it "keeps the first one and does not save the second one" do
-          args = {"tags" => [START_TAG], ID_FIELD => "id123"}
-          first_event = event(args)
-          second_event = event(args)
+      context "if 'keep_start_event' is set to 'last'" do
+        before(:each) do
+          setup_filter("keep_start_event" => 'last')
+        end
 
-          @filter.filter(first_event)
-          @filter.filter(second_event)
+        it "keeps the second one and does not save the first one" do
+            args = {"tags" => [START_TAG], ID_FIELD => "id123"}
+            first_event = event(args)
+            second_event = event(args)
 
-          insist { @filter.start_events.size } == 1
-          insist { @filter.start_events["id123"].event } == first_event
+            @filter.filter(first_event)
+            @filter.filter(second_event)
+
+            insist { @filter.start_events.size } == 1
+            insist { @filter.start_events["id123"].event } == second_event
+        end
+      end
+
+      context "if 'keep_start_event' is set to 'first'" do
+        it "keeps the first one and does not save the second one" do
+            args = {"tags" => [START_TAG], ID_FIELD => "id123"}
+            first_event = event(args)
+            second_event = event(args)
+
+            @filter.filter(first_event)
+            @filter.filter(second_event)
+
+            insist { @filter.start_events.size } == 1
+            insist { @filter.start_events["id123"].event } == first_event
+        end
       end
     end
   end