ES|QL PoC for es-input.

Author: mashhurs

lib/logstash/inputs/elasticsearch.rb

@@ -74,6 +74,7 @@ class LogStash::Inputs::Elasticsearch < LogStash::Inputs::Base
   require 'logstash/inputs/elasticsearch/paginated_search'
   require 'logstash/inputs/elasticsearch/aggregation'
   require 'logstash/inputs/elasticsearch/cursor_tracker'
+  require 'logstash/inputs/elasticsearch/esql'
 
   include LogStash::PluginMixins::ECSCompatibilitySupport(:disabled, :v1, :v8 => :v1)
   include LogStash::PluginMixins::ECSCompatibilitySupport::TargetCheck
@@ -125,20 +126,6 @@ class LogStash::Inputs::Elasticsearch < LogStash::Inputs::Base
   # by this pipeline input.
   config :slices, :validate => :number
 
-  # Enable tracking the value of a given field to be used as a cursor
-  # Main concerns:
-  #       * using anything other than _event.timestamp easily leads to data loss
-  #       * the first "synchronization run can take a long time"
-  config :tracking_field, :validate => :string
-
-  # Define the initial seed value of the tracking_field
-  config :tracking_field_seed, :validate => :string, :default => "1970-01-01T00:00:00.000000000Z"
-
-  # The location of where the tracking field value will be stored
-  # The value is persisted after each scheduled run (and not per result)
-  # If it's not set it defaults to '${path.data}/plugins/inputs/elasticsearch/<pipeline_id>/last_run_value'
-  config :last_run_metadata_path, :validate => :string
-
   # If set, include Elasticsearch document information such as index, type, and
   # the id in the event.
   #
@@ -265,10 +252,6 @@ class LogStash::Inputs::Elasticsearch < LogStash::Inputs::Base
   # exactly once.
   config :schedule, :validate => :string
 
-  # 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 => :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
 
@@ -347,30 +330,16 @@ def register
 
     setup_query_executor
 
-    setup_cursor_tracker
-
     @client
   end
 
   def run(output_queue)
     if @schedule
-      scheduler.cron(@schedule, :overlap => @schedule_overlap) do
-        @query_executor.do_run(output_queue, get_query_object())
-      end
+      scheduler.cron(@schedule) { @query_executor.do_run(output_queue) }
       scheduler.join
     else
-      @query_executor.do_run(output_queue, get_query_object())
-    end
-  end
-
-  def get_query_object
-    if @cursor_tracker
-      query = @cursor_tracker.inject_cursor(@query)
-      @logger.debug("new query is #{query}")
-    else
-      query = @query
+      @query_executor.do_run(output_queue)
     end
-    LogStash::Json.load(query)
   end
 
   ##
@@ -380,11 +349,6 @@ def push_hit(hit, output_queue, root_field = '_source')
     event = event_from_hit(hit, root_field)
     decorate(event)
     output_queue << event
-    record_last_value(event)
-  end
-
-  def record_last_value(event)
-    @cursor_tracker.record_last_value(event) if @tracking_field
   end
 
   def event_from_hit(hit, root_field)
@@ -678,28 +642,6 @@ def setup_query_executor
                       end
   end
 
-  def setup_cursor_tracker
-    return unless @tracking_field
-    return unless @query_executor.is_a?(LogStash::Inputs::Elasticsearch::SearchAfter)
-
-    if @resolved_search_api != "search_after" || @response_type != "hits"
-      raise ConfigurationError.new("The `tracking_field` feature can only be used with `search_after` non-aggregation queries")
-    end
-
-    @cursor_tracker = CursorTracker.new(last_run_metadata_path: last_run_metadata_path,
-                                        tracking_field: @tracking_field,
-                                        tracking_field_seed: @tracking_field_seed)
-    @query_executor.cursor_tracker = @cursor_tracker
-  end
-
-  def last_run_metadata_path
-    return @last_run_metadata_path if @last_run_metadata_path
-
-    last_run_metadata_path = ::File.join(LogStash::SETTINGS.get_value("path.data"), "plugins", "inputs", "elasticsearch", pipeline_id, "last_run_value")
-    FileUtils.mkdir_p ::File.dirname(last_run_metadata_path)
-    last_run_metadata_path
-  end
-
   def get_transport_client_class
     # LS-core includes `elasticsearch` gem. The gem is composed of two separate gems: `elasticsearch-api` and `elasticsearch-transport`
     # And now `elasticsearch-transport` is old, instead we have `elastic-transport`.

lib/logstash/inputs/elasticsearch/esql.rb

@@ -0,0 +1,67 @@
+require 'logstash/helpers/loggable_try'
+
+module LogStash
+  module Inputs
+    class Elasticsearch
+      class Esql
+        include LogStash::Util::Loggable
+
+        ESQL_JOB = "ES|QL job"
+
+        def initialize(client, plugin)
+          @client = client
+          @plugin_params = plugin.params
+          @plugin = plugin
+          @retries = @plugin_params["retries"]
+
+          @query = @plugin_params["query"]
+          esql_options = @plugin_params["esql"] ? @plugin_params["esql"]: {}
+          @esql_params = esql_options["params"] ? esql_options["params"] : {}
+          # TODO: add filter as well
+          # @esql_params = esql_options["filter"] | []
+        end
+
+        def retryable(job_name, &block)
+          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("#{job_name} failed with ", error_details)
+          false
+        end
+
+        def do_run(output_queue)
+          logger.info("ES|QL executor starting")
+          response = retryable(ESQL_JOB) do
+            @client.esql.query({ body: { query: @query }, format: 'json' })
+            # TODO: make sure to add params, filters, etc...
+            # @client.esql.query({ body: { query: @query }, format: 'json' }.merge(@esql_params))
+
+          end
+          puts "Response: #{response.inspect}"
+          if response && response['values']
+            response['values'].each do |value|
+              mapped_data = map_column_and_values(response['columns'], value)
+              puts "Mapped Data: #{mapped_data}"
+              @plugin.decorate_and_push_to_queue(output_queue, mapped_data)
+            end
+          end
+        end
+
+        def map_column_and_values(columns, values)
+          puts "columns class: #{columns.class}"
+          puts "values class: #{values.class}"
+          puts "columns: #{columns.inspect}"
+          puts "values: #{values.inspect}"
+          mapped_data = {}
+          columns.each_with_index do |column, index|
+            mapped_data[column["name"]] = values[index]
+          end
+          puts "values: #{mapped_data.inspect}"
+          mapped_data
+        end
+      end
+    end
+  end
+end