Introduce query_params option to accept drop_null_columns, set default timestampt converter to LogStash::Timestamp, dotted fields extended to nested fields.

Author: mashhurs

docs/index.asciidoc

@@ -266,7 +266,13 @@ The following is a basic scheduled ES|QL query that runs hourly:
 
 Set `config.support_escapes: true` in `logstash.yml` if you need to escape special chars in the query.
 
-NOTE: With ES|QL query, {ls} doesn't generate `event.original`
+NOTE: With ES|QL query, {ls} doesn't generate `event.original`.
+
+Consider the following caveat scenarios:
+
+- ES|QL by default returns entire columns even if their values are `null`. The plugin provides a `drop_null_columns` option via <<plugins-{type}s-{plugin}-query_params>>. Enabling this parameter instructs {es} to automatically exclude columns with null values from query results.
+- If your {es} index uses https://www.elastic.co/docs/reference/elasticsearch/mapping-reference/multi-fields[multi-fields] mapping(s), ES|QL query fetches all parent and sub-fields fields. Since {ls} events cannot contain parent field's concrete value and sub-field values together, we recommend using the `DROP` keyword in your ES|QL query explicitly remove sub-fields.
+- If your {es} index contains top level `tags` field, this will conflict with {ls} event's reserved `tags` field. {ls} moves `tags` field values to the `_tags` and populates `tags` with `["_tagsparsefailure"]`.
 
 For comprehensive ES|QL syntax reference and best practices, see the https://www.elastic.co/guide/en/elasticsearch/reference/current/esql-syntax.html[official {es} documentation].
 
@@ -297,6 +303,7 @@ Please check out <<plugins-{type}s-{plugin}-obsolete-options>> for details.
 | <<plugins-{type}s-{plugin}-password>> |<<password,password>>|No
 | <<plugins-{type}s-{plugin}-proxy>> |<<uri,uri>>|No
 | <<plugins-{type}s-{plugin}-query>> |<<string,string>>|No
+| <<plugins-{type}s-{plugin}-query_params>> |<<hash,hash>>|No
 | <<plugins-{type}s-{plugin}-response_type>> |<<string,string>>, one of `["hits","aggregations","esql"]`|No
 | <<plugins-{type}s-{plugin}-request_timeout_seconds>> | <<number,number>>|No
 | <<plugins-{type}s-{plugin}-schedule>> |<<string,string>>|No
@@ -544,6 +551,31 @@ 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}-query_params"]
+===== `query_params`
+Parameters to send to {es} together with <<plugins-{type}s-{plugin}-query>>.
+
+Accepted options:
+[cols="2,1,3",options="header"]
+|===
+|Option name |Default value | Description
+
+|`drop_null_columns` |`false` | Requests {es} to filter out `null` columns
+|===
+
+Example
+[source, ruby]
+    input {
+      elasticsearch {
+        response_type => 'esql'
+        query => 'FROM access-logs* | WHERE type="apache"'
+        query_params => {
+          drop_null_columns => true
+        }
+      }
+    }
+
+
 [id="plugins-{type}s-{plugin}-response_type"]
 ===== `response_type`
 

lib/logstash/inputs/elasticsearch.rb

@@ -276,6 +276,10 @@ 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
 
+  # Parameters query or query APIs can use
+  #   current acceptable params: drop_null_columns => true|false (for ES|QL)
+  config :query_params, :validate => :hash, :default => {}
+
   # Obsolete Settings
   config :ssl, :obsolete => "Set 'ssl_enabled' instead."
   config :ca_file, :obsolete => "Set 'ssl_certificate_authorities' instead."
@@ -323,6 +327,7 @@ def register
 
     @retries < 0 && fail(LogStash::ConfigurationError, "Elasticsearch Input Plugin's `retries` option must be equal or greater than zero, got `#{@retries}`")
 
+    validate_query_params!
     validate_authentication
     fill_user_password_from_cloud_auth
 
@@ -751,6 +756,17 @@ def validate_esql_query!
     fail(LogStash::ConfigurationError, "`query` needs to start with any of #{source_commands}") unless contains_source_command
   end
 
+  def validate_query_params!
+    # keep the original, remove ES|QL accepted params and validate
+    cloned_query_params = @query_params.clone
+    if @response_type == 'esql'
+      cloned_query_params.delete("drop_null_columns")
+      fail(LogStash::ConfigurationError, "#{cloned_query_params} not accepted when `response_type => 'esql'`") if cloned_query_params.any?
+    else
+      fail(LogStash::ConfigurationError, "#{@query_params} not accepted when `response_type => #{@response_type}`") if @query_params.any?
+    end
+  end
+
   def inform_ineffective_esql_params
     ineffective_options = original_params.keys & %w(index target size slices search_api, docinfo, docinfo_target, docinfo_fields)
     @logger.info("Configured #{ineffective_options} params are ineffective in ES|QL mode") if ineffective_options.size > 1

lib/logstash/inputs/elasticsearch/esql.rb

@@ -8,6 +8,10 @@ class Esql
 
         ESQL_JOB = "ES|QL job"
 
+        ESQL_PARSERS_BY_TYPE = Hash.new(lambda { |x| x }).merge(
+          'date' => ->(value) { value && LogStash::Timestamp.new(value) },
+          )
+
         # Initialize the ESQL query executor
         # @param client [Elasticsearch::Client] The Elasticsearch client instance
         # @param plugin [LogStash::Inputs::Elasticsearch] The parent plugin instance
@@ -20,6 +24,9 @@ def initialize(client, plugin)
           unless @query.include?('METADATA')
             logger.warn("The query doesn't have METADATA keyword. Including it makes _id and _version available in the documents", {:query => @query})
           end
+
+          params = plugin.params["query_params"] || {}
+          @drop_null_columns = params["drop_null_columns"] || false
         end
 
         # Execute the ESQL query and process results
@@ -28,7 +35,7 @@ def initialize(client, plugin)
         def do_run(output_queue, query)
           logger.info("ES|QL executor starting")
           response = retryable(ESQL_JOB) do
-            @client.esql.query({ body: { query: @query }, format: 'json' })
+            @client.esql.query({ body: { query: @query }, format: 'json', drop_null_columns: @drop_null_columns })
           end
           # retriable already printed error details
           return if response == false
@@ -64,7 +71,13 @@ def retryable(job_name, &block)
         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)
+            nest_structured_data = nest_keys(mapped_data)
+            @plugin.decorate_and_push_to_queue(output_queue, nest_structured_data)
+          rescue => e
+            # if event creation fails with whatever reason, inform user and tag with failure and return entry as it is
+            logger.warn("Event creation error, ", message: e.message, exception: e.class, data: { "columns" => columns, "values" => [value] })
+            failed_event = LogStash::Event.new("columns" => columns, "values" => [value], "tags" => ['_elasticsearch_input_failure'])
+            output_queue << failed_event
           end
         end
 
@@ -74,7 +87,19 @@ def process_response(values, columns, output_queue)
         # @return [Hash] Mapped data with column names as keys
         def map_column_and_values(columns, values)
           columns.each_with_index.with_object({}) do |(column, index), mapped_data|
-            mapped_data[column["name"]] = values[index]
+            mapped_data[column["name"]] = ESQL_PARSERS_BY_TYPE[column["type"]].call(values[index])
+          end
+        end
+
+        # Transforms dotted keys to nested JSON shape
+        # @param dot_keyed_hash [Hash] whose keys are dotted (example 'a.b.c.d': 'val')
+        # @return [Hash] whose keys are nested with value mapped ({'a':{'b':{'c':{'d':'val'}}}})
+        def nest_keys(dot_keyed_hash)
+          dot_keyed_hash.each_with_object({}) do |(key, value), result|
+            key_parts = key.to_s.split('.')
+            *path, leaf = key_parts
+            leaf_scope = path.inject(result) { |scope, part| scope[part] ||= {} }
+            leaf_scope[leaf] = value
           end
         end
       end

spec/inputs/elasticsearch_esql_spec.rb

@@ -44,7 +44,7 @@
 
   describe "when executing chain of processes" do
     let(:output_queue) { Queue.new }
-    let(:response) { { 'values' => [%w[foo bar]], 'columns' => [{ 'name' => 'id'}, { 'name' => 'val'}] } }
+    let(:response) { { 'values' => [%w[foo bar]], 'columns' => [{ 'name' => 'a.b.1.d'}, { 'name' => 'h_g.k$l.m.0'}] } }
 
     before do
       allow(esql_executor).to receive(:retryable).and_yield
@@ -55,7 +55,7 @@
     it "executes the ESQL query and processes the results" do
       allow(response).to receive(:headers).and_return({})
       esql_executor.do_run(output_queue, plugin_config["query"])
-      expect(plugin).to have_received(:decorate_and_push_to_queue).with(output_queue, {'id' => 'foo', 'val' => 'bar'})
+      expect(plugin).to have_received(:decorate_and_push_to_queue).with(output_queue, {"a"=>{"b"=>{"1"=>{"d"=>"foo"}}}, "h_g"=>{"k$l"=>{"m"=>{"0"=>"bar"}}}})
     end
 
     it "logs a warning if the response contains a warning header" do
@@ -71,16 +71,15 @@
     end
   end
 
-
   describe "when starts processing the response" do
     let(:output_queue) { Queue.new }
     let(:values) { [%w[foo bar]] }
-    let(:columns) { [{'name' => 'id'}, {'name' => 'val'}] }
+    let(:columns) { [{'name' => 'some.id'}, {'name' => 'some.val'}] }
 
     it "processes the ESQL response and pushes events to the output queue" do
       allow(plugin).to receive(:decorate_and_push_to_queue)
       esql_executor.send(:process_response, values, columns, output_queue)
-      expect(plugin).to have_received(:decorate_and_push_to_queue).with(output_queue, {'id' => 'foo', 'val' => 'bar'})
+      expect(plugin).to have_received(:decorate_and_push_to_queue).with(output_queue, {"some"=>{"id"=>"foo", "val"=>"bar"}})
     end
   end
 

spec/inputs/elasticsearch_spec.rb

@@ -1473,6 +1473,38 @@ def extract_transport(client) # on 7.x client.transport is a ES::Transport::Clie
           end
         end
       end
+
+      describe "with extra params" do
+        context "empty `query_params`" do
+          let(:config) {
+            super().merge('query_params' => {})
+          }
+
+          it "does not raise a configuration error" do
+            expect { plugin.send(:validate_query_params!) }.not_to raise_error
+          end
+        end
+
+        context "with actual `drop_null_columns` value" do
+          let(:config) {
+            super().merge('query_params' => { 'drop_null_columns' => true })
+          }
+
+          it "does not raise a configuration error" do
+            expect { plugin.send(:validate_query_params!) }.not_to raise_error
+          end
+        end
+
+        context "with extra non ES|QL params" do
+          let(:config) {
+            super().merge('query_params' => { 'drop_null_columns' => true, 'test' => 'hi'})
+          }
+
+          it "does not raise a configuration error" do
+            expect { plugin.send(:validate_query_params!) }.to raise_error(LogStash::ConfigurationError, "{\"test\"=>\"hi\"} not accepted when `response_type => 'esql'`")
+          end
+        end
+      end
     end
   end
 end