Add support to search after (#198)

This commit introduces the `search_api` option to enable support for both search_after and scroll.

`auto`: Default value. It uses `search_after` for Elasticsearch 8 and newer versions, otherwise it falls back to the `scroll` API
`search_after`: Create a point in time for search, which is the recommended way to do paginated search since 7.10
`scroll`: Uses scroll API to search, which is the previous way to search

When search_after is utilized, if the query doesn't specify a sort field, a default sort of { "sort": { "_shard_doc": "asc" } } will be added to the query. By default, Elasticsearch adds a sort field "_shard_doc" on top of existing sort implicitly as a tie-breakers.

The scroll search has been refactored and is expected to function the same, with only minor changes made to the logging messages.

Co-authored-by: João Duarte <[email protected]>

Author: kaisecheng

CHANGELOG.md

@@ -1,3 +1,7 @@
+## 4.19.0
+  - Added `search_api` option to support `search_after` and `scroll` [#198](https://github.com/logstash-plugins/logstash-input-elasticsearch/pull/198)
+    - The default value `auto` uses `search_after` for Elasticsearch >= 8, otherwise, fall back to `scroll` 
+
 ## 4.18.0
   - Added request header `Elastic-Api-Version` for serverless [#195](https://github.com/logstash-plugins/logstash-input-elasticsearch/pull/195)
 

docs/index.asciidoc

@@ -118,6 +118,7 @@ This plugin supports the following configuration options plus the <<plugins-{typ
 | <<plugins-{type}s-{plugin}-request_timeout_seconds>> | <<number,number>>|No
 | <<plugins-{type}s-{plugin}-schedule>> |<<string,string>>|No
 | <<plugins-{type}s-{plugin}-scroll>> |<<string,string>>|No
+| <<plugins-{type}s-{plugin}-search_api>> |<<string,string>>, one of `["auto", "search_after", "scroll"]`|No
 | <<plugins-{type}s-{plugin}-size>> |<<number,number>>|No
 | <<plugins-{type}s-{plugin}-slices>> |<<number,number>>|No
 | <<plugins-{type}s-{plugin}-ssl_certificate>> |<<path,path>>|No
@@ -333,6 +334,9 @@ environment variables e.g. `proxy => '${LS_PROXY:}'`.
 The query to be executed. Read the {ref}/query-dsl.html[Elasticsearch query DSL
 documentation] for more information.
 
+When <<plugins-{type}s-{plugin}-search_api>> resolves to `search_after` and the query does not specify `sort`,
+the default sort `'{ "sort": { "_shard_doc": "asc" } }'` will be added to the query. Please refer to the {ref}/paginate-search-results.html#search-after[Elasticsearch search_after] parameter to know more.
+
 [id="plugins-{type}s-{plugin}-request_timeout_seconds"]
 ===== `request_timeout_seconds`
 
@@ -377,6 +381,19 @@ This parameter controls the keepalive time in seconds of the scrolling
 request and initiates the scrolling process. The timeout applies per
 round trip (i.e. between the previous scroll request, to the next).
 
+[id="plugins-{type}s-{plugin}-seearch_api"]
+===== `search_api`
+
+* Value can be any of: `auto`, `search_after`, `scroll`
+* Default value is `auto`
+
+With `auto` the plugin uses the `search_after` parameter for Elasticsearch version `8.0.0` or higher, otherwise the `scroll` API is used instead.
+
+`search_after` uses {ref}/point-in-time-api.html#point-in-time-api[point in time] and sort value to search.
+The query requires at least one `sort` field, as described in the <<plugins-{type}s-{plugin}-query>> parameter.
+
+`scroll` uses {ref}/paginate-search-results.html#scroll-search-results[scroll] API to search, which is no longer recommended.
+
 [id="plugins-{type}s-{plugin}-size"]
 ===== `size` 
 

lib/logstash/inputs/elasticsearch.rb

@@ -11,7 +11,6 @@
 require "logstash/plugin_mixins/scheduler"
 require "logstash/plugin_mixins/normalize_config_support"
 require "base64"
-require 'logstash/helpers/loggable_try'
 
 require "elasticsearch"
 require "elasticsearch/transport/transport/http/manticore"
@@ -74,6 +73,8 @@
 #
 class LogStash::Inputs::Elasticsearch < LogStash::Inputs::Base
 
+  require 'logstash/inputs/elasticsearch/paginated_search'
+
   include LogStash::PluginMixins::ECSCompatibilitySupport(:disabled, :v1, :v8 => :v1)
   include LogStash::PluginMixins::ECSCompatibilitySupport::TargetCheck
 
@@ -106,6 +107,10 @@ class LogStash::Inputs::Elasticsearch < LogStash::Inputs::Base
   # The number of retries to run the query. If the query fails after all retries, it logs an error message.
   config :retries, :validate => :number, :default => 0
 
+  # Default `auto` will use `search_after` api for Elasticsearch 8 and use `scroll` api for 7
+  # Set to scroll to fallback to previous version
+  config :search_api, :validate => %w[auto search_after scroll], :default => "auto"
+
   # This parameter controls the keepalive time in seconds of the scrolling
   # request and initiates the scrolling process. The timeout applies per
   # round trip (i.e. between the previous scroll request, to the next).
@@ -321,93 +326,21 @@ def register
 
     setup_serverless
 
+    setup_search_api
+
     @client
   end
 
 
   def run(output_queue)
     if @schedule
-      scheduler.cron(@schedule) { do_run(output_queue) }
+      scheduler.cron(@schedule) { @paginated_search.do_run(output_queue) }
       scheduler.join
     else
-      do_run(output_queue)
-    end
-  end
-
-  private
-  JOB_NAME = "run query"
-  def do_run(output_queue)
-    # if configured to run a single slice, don't bother spinning up threads
-    if @slices.nil? || @slices <= 1
-      return retryable(JOB_NAME) do
-        do_run_slice(output_queue)
-      end
-    end
-
-    logger.warn("managed slices for query is very large (#{@slices}); consider reducing") if @slices > 8
-
-
-    @slices.times.map do |slice_id|
-      Thread.new do
-        LogStash::Util::set_thread_name("[#{pipeline_id}]|input|elasticsearch|slice_#{slice_id}")
-        retryable(JOB_NAME) do
-          do_run_slice(output_queue, slice_id)
-        end
-      end
-    end.map(&:join)
-
-    logger.trace("#{@slices} slices completed")
-  end
-
-  def retryable(job_name, &block)
-    begin
-      stud_try = ::LogStash::Helpers::LoggableTry.new(logger, job_name)
-      stud_try.try((@retries + 1).times) { yield }
-    rescue => e
-      error_details = {:message => e.message, :cause => e.cause}
-      error_details[:backtrace] = e.backtrace if logger.debug?
-      logger.error("Tried #{job_name} unsuccessfully", error_details)
+      @paginated_search.do_run(output_queue)
     end
   end
 
-  def do_run_slice(output_queue, slice_id=nil)
-    slice_query = @base_query
-    slice_query = slice_query.merge('slice' => { 'id' => slice_id, 'max' => @slices}) unless slice_id.nil?
-
-    slice_options = @options.merge(:body => LogStash::Json.dump(slice_query) )
-
-    logger.info("Slice starting", slice_id: slice_id, slices: @slices) unless slice_id.nil?
-
-    begin
-      r = search_request(slice_options)
-
-      r['hits']['hits'].each { |hit| push_hit(hit, output_queue) }
-      logger.debug("Slice progress", slice_id: slice_id, slices: @slices) unless slice_id.nil?
-
-      has_hits = r['hits']['hits'].any?
-      scroll_id = r['_scroll_id']
-
-      while has_hits && scroll_id && !stop?
-        has_hits, scroll_id = process_next_scroll(output_queue, scroll_id)
-        logger.debug("Slice progress", slice_id: slice_id, slices: @slices) if logger.debug? && slice_id
-      end
-      logger.info("Slice complete", slice_id: slice_id, slices: @slices) unless slice_id.nil?
-    ensure
-      clear_scroll(scroll_id)
-    end
-  end
-
-  ##
-  # @param output_queue [#<<]
-  # @param scroll_id [String]: a scroll id to resume
-  # @return [Array(Boolean,String)]: a tuple representing whether the response
-  #
-  def process_next_scroll(output_queue, scroll_id)
-    r = scroll_request(scroll_id)
-    r['hits']['hits'].each { |hit| push_hit(hit, output_queue) }
-    [r['hits']['hits'].any?, r['_scroll_id']]
-  end
-
   def push_hit(hit, output_queue)
     event = targeted_event_factory.new_event hit['_source']
     set_docinfo_fields(hit, event) if @docinfo
@@ -433,20 +366,7 @@ def set_docinfo_fields(hit, event)
     event.set(@docinfo_target, docinfo_target)
   end
 
-  def clear_scroll(scroll_id)
-    @client.clear_scroll(:body => { :scroll_id => scroll_id }) if scroll_id
-  rescue => e
-    # ignore & log any clear_scroll errors
-    logger.warn("Ignoring clear_scroll exception", message: e.message, exception: e.class)
-  end
-
-  def scroll_request(scroll_id)
-    @client.scroll(:body => { :scroll_id => scroll_id }, :scroll => @scroll)
-  end
-
-  def search_request(options={})
-    @client.search(options)
-  end
+  private
 
   def hosts_default?(hosts)
     hosts.nil? || ( hosts.is_a?(Array) && hosts.empty? )
@@ -677,6 +597,18 @@ def test_connection!
     raise LogStash::ConfigurationError, "Could not connect to a compatible version of Elasticsearch"
   end
 
+  def es_info
+    @es_info ||= @client.info
+  end
+
+  def es_version
+    @es_version ||= es_info&.dig('version', 'number')
+  end
+
+  def es_major_version
+    @es_major_version ||= es_version.split('.').first.to_i
+  end
+
   # recreate client with default header when it is serverless
   # verify the header by sending GET /
   def setup_serverless
@@ -691,13 +623,35 @@ def setup_serverless
   end
 
   def build_flavor
-    @build_flavor ||= @client.info&.dig('version', 'build_flavor')
+    @build_flavor ||= es_info&.dig('version', 'build_flavor')
   end
 
   def serverless?
     @is_serverless ||= (build_flavor == BUILD_FLAVOR_SERVERLESS)
   end
 
+  def setup_search_api
+    @resolved_search_api = if @search_api == "auto"
+                             api = if es_major_version >= 8
+                                    "search_after"
+                                   else
+                                     "scroll"
+                                   end
+                             logger.info("`search_api => auto` resolved to `#{api}`", :elasticsearch => es_version)
+                             api
+                           else
+                             @search_api
+                           end
+
+
+    @paginated_search = if @resolved_search_api == "search_after"
+                          LogStash::Inputs::Elasticsearch::SearchAfter.new(@client, self)
+                        else
+                          logger.warn("scroll API is no longer recommended for pagination. Consider using search_after instead.") if es_major_version >= 8
+                          LogStash::Inputs::Elasticsearch::Scroll.new(@client, self)
+                        end
+  end
+
   module URIOrEmptyValidator
     ##
     # @override to provide :uri_or_empty validator