ES|QL support: ESQL executor implementation, response type to accept esql option, validations to make sure both LS and ES support the ESQL execution.

Author: mashhurs

CHANGELOG.md

@@ -1,3 +1,6 @@
+## 5.1.0
+  - ES|QL support [#N](https://github.com/logstash-plugins/logstash-input-elasticsearch/pull/N)
+
 ## 5.0.2
   - Add elastic-transport client support used in elasticsearch-ruby 8.x [#223](https://github.com/logstash-plugins/logstash-input-elasticsearch/pull/223)
 

lib/logstash/inputs/elasticsearch.rb

@@ -96,15 +96,17 @@ class LogStash::Inputs::Elasticsearch < LogStash::Inputs::Base
   # The index or alias to search.
   config :index, :validate => :string, :default => "logstash-*"
 
-  # The query to be executed. Read the Elasticsearch query DSL documentation
-  # for more info
-  # https://www.elastic.co/guide/en/elasticsearch/reference/current/query-dsl.html
+  # The query to be executed.
+  # It accepts DSL and ES|QL query shapes.
+  # Read the Elasticsearch query documentations for more info:
+  # - DSL: https://www.elastic.co/guide/en/elasticsearch/reference/current/query-dsl.html
+  # - ES|QL: https://www.elastic.co/guide/en/elasticsearch/reference/current/esql.html
   config :query, :validate => :string, :default => '{ "sort": [ "_doc" ] }'
 
   # This allows you to speccify the response type: either hits or aggregations
   # where hits: normal search request
   #       aggregations: aggregation request
-  config :response_type, :validate => ['hits', 'aggregations'], :default => 'hits'
+  config :response_type, :validate => %w[hits aggregations esql], :default => 'hits'
 
   # This allows you to set the maximum number of hits returned per scroll.
   config :size, :validate => :number, :default => 1000
@@ -254,11 +256,6 @@ class LogStash::Inputs::Elasticsearch < LogStash::Inputs::Base
   # 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
 
-  # A mode to switch between DSL and ES|QL, defaults to DSL
-  config :query_mode, :validate => %w[dsl, esql], :default => 'dsl'
-
-  config :esql, :validate => :hash
-
   # Obsolete Settings
   config :ssl, :obsolete => "Set 'ssl_enabled' instead."
   config :ca_file, :obsolete => "Set 'ssl_certificate_authorities' instead."
@@ -289,8 +286,12 @@ def register
     fill_hosts_from_cloud_id
     setup_ssl_params!
 
-    puts "Query mode: #{@query_mode}"
-    if @query_mode == 'dsl'
+    if @response_type == 'esql'
+      validate_ls_version_for_esql_support!
+      validate_esql_query!
+      inform_ineffective_esql_params
+    else
+      # for the ES|QL, plugin accepts raw string query but JSON for others
       @base_query = LogStash::Json.load(@query)
       if @slices
         @base_query.include?('slice') && fail(LogStash::ConfigurationError, "Elasticsearch Input Plugin's `query` option cannot specify specific `slice` when configured to manage parallel slices with `slices` option")
@@ -331,6 +332,9 @@ def register
 
     test_connection!
 
+    # make sure connected ES supports ES|QL (8.11+)
+    validate_es_for_esql_support! if @response_type == 'esql'
+
     setup_serverless
 
     setup_search_api
@@ -369,10 +373,7 @@ def event_from_hit(hit, root_field)
     return event_factory.new_event('event' => { 'original' => serialized_hit }, 'tags' => ['_elasticsearch_input_failure'])
   end
 
-  # hit: {columns: [], values: []}
   def decorate_and_push_to_queue(output_queue, mapped_entry)
-    puts "mapped_entry class: #{mapped_entry.class}"
-    puts "mapped_entry value: #{mapped_entry.inspect}"
     event = targeted_event_factory.new_event mapped_entry
     decorate(event)
     output_queue << event
@@ -645,11 +646,6 @@ def setup_search_api
   end
 
   def setup_query_executor
-    if @query_mode == 'esql'
-      @query_executor = LogStash::Inputs::Elasticsearch::Esql.new(@client, self)
-      return @query_executor
-    end
-
     @query_executor = case @response_type
                       when 'hits'
                         if @resolved_search_api == "search_after"
@@ -660,6 +656,8 @@ def setup_query_executor
                         end
                       when 'aggregations'
                         LogStash::Inputs::Elasticsearch::Aggregation.new(@client, self)
+                      when 'esql'
+                        LogStash::Inputs::Elasticsearch::Esql.new(@client, self)
                       end
   end
 
@@ -677,6 +675,31 @@ def get_transport_client_class
     ::Elastic::Transport::Transport::HTTP::Manticore
   end
 
+  def validate_ls_version_for_esql_support!
+    # LS 8.17.4+ has elasticsearch-ruby 8.17 client
+    # elasticsearch-ruby 8.11+ supports ES|QL
+    if Gem::Version.create(LOGSTASH_VERSION) < Gem::Version.create("8.17.4")
+      fail("Current version of Logstash does not include Elasticsearch client which supports ES|QL. Please upgrade Logstash to at least 8.17.4")
+    end
+  end
+
+  def validate_esql_query!
+    fail(LogStash::ConfigurationError, "`query` cannot be empty") if @query.strip.empty?
+    source_commands = %w[FROM ROW SHOW]
+    contains_source_command = source_commands.any? { |source_command| @query.strip.start_with?(source_command) }
+    fail(LogStash::ConfigurationError, "`query` needs to start with any of #{source_commands}") unless contains_source_command
+  end
+
+  def inform_ineffective_esql_params
+    ineffective_options = original_params.keys & %w(target size slices search_api)
+    @logger.info("Configured #{ineffective_options} params are ineffective in ES|QL mode") if ineffective_options.size > 1
+  end
+
+  def validate_es_for_esql_support!
+    es_supports_esql = Gem::Version.create(es_version) >= Gem::Version.create("8.11")
+    fail("Connected Elasticsearch #{es_version} version does not supports ES|QL. Please upgrade it.") unless es_supports_esql
+  end
+
   module URIOrEmptyValidator
     ##
     # @override to provide :uri_or_empty validator

lib/logstash/inputs/elasticsearch/esql.rb

@@ -8,19 +8,21 @@ class Esql
 
         ESQL_JOB = "ES|QL job"
 
+        # Initialize the ESQL query executor
+        # @param client [Elasticsearch::Client] The Elasticsearch client instance
+        # @param plugin [LogStash::Inputs::Elasticsearch] The parent plugin instance
         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
 
+        # Execute a retryable operation with proper error handling
+        # @param job_name [String] Name of the job for logging purposes
+        # @yield The block to execute
+        # @return [Boolean] true if successful, false otherwise
         def retryable(job_name, &block)
           stud_try = ::LogStash::Helpers::LoggableTry.new(logger, job_name)
           stud_try.try((@retries + 1).times) { yield }
@@ -31,35 +33,47 @@ def retryable(job_name, &block)
           false
         end
 
+        # Execute the ESQL query and process results
+        # @param output_queue [Queue] The queue to push processed events to
         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
+          # retriable already printed error details
+          return if response == false
 
+          puts "response class: #{response.class}"
+          puts "response: #{response.inspect}"
+          unless response&.headers&.dig("warning")
+            logger.warn("ES|QL executor received warning", {:message => response.headers["warning"]})
+          end
+          if response['values'] && response['columns']
+            process_response(response['values'], response['columns'], output_queue)
           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
+
+        private
+
+        # Process the ESQL response and push events to the output queue
+        # @param values [Array[Array]] The ESQL query response hits
+        # @param columns [Array[Hash]] The ESQL query response columns
+        # @param output_queue [Queue] The queue to push processed events to
+        def process_response(values, columns, output_queue)
+          values.each do |value|
+            mapped_data = map_column_and_values(columns, value)
+            @plugin.decorate_and_push_to_queue(output_queue, mapped_data)
           end
         end
 
+        # Map column names to their corresponding values
+        # @param columns [Array] Array of column definitions
+        # @param values [Array] Array of values for the current row
+        # @return [Hash] Mapped data with column names as keys
         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|
+          columns.each_with_index.with_object({}) do |(column, index), mapped_data|
             mapped_data[column["name"]] = values[index]
           end
-          puts "values: #{mapped_data.inspect}"
-          mapped_data
         end
       end
     end

logstash-input-elasticsearch.gemspec

@@ -1,7 +1,7 @@
 Gem::Specification.new do |s|
 
   s.name            = 'logstash-input-elasticsearch'
-  s.version         = '5.0.2'
+  s.version         = '5.1.0'
   s.licenses        = ['Apache License (2.0)']
   s.summary         = "Reads query results from an Elasticsearch cluster"
   s.description     = "This gem is a Logstash plugin required to be installed on top of the Logstash core pipeline using $LS_HOME/bin/logstash-plugin install gemname. This gem is not a stand-alone program"

spec/inputs/elasticsearch_esql_spec.rb

@@ -0,0 +1,115 @@
+# encoding: utf-8
+require "logstash/devutils/rspec/spec_helper"
+require "logstash/inputs/elasticsearch"
+require "elasticsearch"
+require 'logstash/plugin_mixins/ecs_compatibility_support/spec_helper'
+
+describe LogStash::Inputs::Elasticsearch, :ecs_compatibility_support do
+  let(:plugin) { described_class.new(config) }
+  let(:client) { instance_double(Elasticsearch::Client) }
+  let(:queue) { Queue.new }
+  let(:cluster_info) { { "version" => { "number" => "8.11.0", "build_flavor" => "default" } } }
+  
+  let(:config) do
+    {
+      "query" => "FROM test-index | STATS count() BY field",
+      "response_type" => "esql",
+      "retries" => 3
+    }
+  end
+
+  describe "#initialize" do
+    it "sets up the ESQL client with correct parameters" do
+      expect(plugin.instance_variable_get(:@query)).to eq(config["query"])
+      expect(plugin.instance_variable_get(:@response_type)).to eq(config["response_type"])
+      expect(plugin.instance_variable_get(:@retries)).to eq(config["retries"])
+    end
+  end
+
+  describe "#register" do
+    before(:each) do
+      Elasticsearch::Client.send(:define_method, :ping) { }
+      allow_any_instance_of(Elasticsearch::Client).to receive(:info).and_return(cluster_info)
+    end
+    it "creates ES|QL executor" do
+      plugin.register
+      expect(plugin.instance_variable_get(:@query_executor)).to be_an_instance_of(LogStash::Inputs::Elasticsearch::Esql)
+    end
+  end
+
+  describe "#validation" do
+
+    describe "LS version" do
+      context "when compatible" do
+        before(:each) do
+          stub_const("LogStash::VERSION", "8.11.0")
+        end
+
+        it "does not raise an error" do
+          expect { plugin.send(:validate_ls_version_for_esql_support!) }.not_to raise_error
+        end
+      end
+
+      context "when incompatible" do
+        before(:each) do
+          stub_const("LOGSTASH_VERSION", "8.10.0")
+        end
+
+        it "raises a runtime error" do
+          expect { plugin.send(:validate_ls_version_for_esql_support!) }
+            .to raise_error(RuntimeError, /Current version of Logstash does not include Elasticsearch client which supports ES|QL. Please upgrade Logstash to at least 8.17.4/)
+        end
+      end
+    end
+
+    describe "ES version" do
+      before(:each) do
+        allow(plugin).to receive(:es_version).and_return("8.10.5")
+      end
+
+      context "when incompatible" do
+        it "raises a runtime error" do
+          expect { plugin.send(:validate_es_for_esql_support!) }
+            .to raise_error(RuntimeError, /Connected Elasticsearch 8.10.5 version does not supports ES|QL. Please upgrade it./)
+        end
+      end
+    end
+
+    describe "ES|QL query"  do
+      context "when query is valid" do
+        it "does not raise an error" do
+          expect { plugin.send(:validate_esql_query!) }.not_to raise_error
+        end
+      end
+
+      context "when query is empty" do
+        let(:config) do
+          {
+            "query" => " "
+          }
+        end
+
+        # TODO: make shared spec
+        it "raises a configuration error" do
+          expect { plugin.send(:validate_esql_query!) }
+            .to raise_error(LogStash::ConfigurationError, /`query` cannot be empty/)
+        end
+      end
+
+      context "when query doesn't align with ES syntax" do
+        let(:config) do
+          {
+            "query" => "RANDOM query"
+          }
+        end
+
+        it "raises a configuration error" do
+          source_commands = %w[FROM ROW SHOW]
+          expect { plugin.send(:validate_esql_query!) }
+            .to raise_error(LogStash::ConfigurationError, "`query` needs to start with any of #{source_commands}")
+        end
+      end
+    end
+  end
+
+end