add :present injection to safeguard against PIT right-edge inconsistencies

Author: jsvd

lib/logstash/inputs/elasticsearch.rb

@@ -129,7 +129,6 @@ class LogStash::Inputs::Elasticsearch < LogStash::Inputs::Base
 
   # Enable tracking the value of a given field to be used as a cursor
   # TODO: main concerns
-  #       * schedule overlap needs to be disabled (hardcoded as enabled)
   #       * using anything other than _event.timestamp easily leads to data loss
   #       * the first "synchronization run can take a long time"
   #       * checkpointing is only safe to do after each run (not per document)
@@ -271,7 +270,7 @@ class LogStash::Inputs::Elasticsearch < LogStash::Inputs::Base
 
   # Allow scheduled runs to overlap (enabled by default). Setting to false will
   # only start a new scheduled run after the previous one completes.
-  config :schedule_overlap, :validate => :string
+  config :schedule_overlap, :validate => :boolean
 
   # If set, the _source of each hit will be added nested under the target instead of at the top-level
   config :target, :validate => :field_reference

lib/logstash/inputs/elasticsearch/cursor_tracker.rb

@@ -35,8 +35,8 @@ def checkpoint_cursor
 
     def converge_last_value
       return if @last_value_hashmap.empty?
-      # TODO this implicitly assumes that the way to converge the value among slices is to pick the highest and we can't assume that
-      new_last_value = @last_value_hashmap.reduceValues(1, lambda { |v1, v2| Time.parse(v1) < Time.parse(v2) ? v2 : v1 })
+      new_last_value = @last_value_hashmap.reduceValues(1000, lambda { |v1, v2| Java::java.time.Instant.parse(v1).isBefore(Java::java.time.Instant.parse(v2)) ? v2 : v1 })
+      logger.trace? && logger.trace("converge_last_value: got #{@last_value_hashmap.values.inspect}. won: #{new_last_value}")
       return if new_last_value == @last_value
       @last_value = new_last_value
       logger.info "New cursor value for field \"#{@tracking_field}\" is: #{new_last_value}"
@@ -49,7 +49,10 @@ def record_last_value(event)
     end
 
     def inject_cursor(query_json)
-      query_json.gsub(":last_value", @last_value.to_s)
+      # ":present" means "now - 30s" to avoid grabbing partially visible data in the PIT
+      result = query_json.gsub(":last_value", @last_value.to_s).gsub(":present", Java::java.time.Instant.now.minusSeconds(30).to_s)
+      logger.debug("inject_cursor: injected values for ':last_value' and ':present'", :query => result)
+      result
     end
   end
 end; end; end