Rebase against upstream main after target support added. Separate unit test for DSL. Address comments: do not save ES version in client, add apply target method in executors, set to target if target is defined, docs update.

Co-authored-by: Rye Biesemeyer <[email protected]>

Author: mashhurs

docs/index.asciidoc

@@ -375,10 +375,11 @@ environment variables e.g. `proxy => '${LS_PROXY:}'`.
   * Value type is <<string,string>>
   * There is no default value for this setting.
 
-Elasticsearch query string. More information is available in the
-{ref}/query-dsl-query-string-query.html#query-string-syntax[Elasticsearch query
-string documentation].
-Use either `query` or `query_template`.
+The query to be executed.
+Accepted query shape is DSL query string or ES|QL.
+For the DSL query string, use either `query` or `query_template`.
+Read the {ref}/query-dsl-query-string-query.html[{es} query
+string documentation] or {ref}/esql.html[{es} ES|QL documentation] for more information.
 
 [id="plugins-{type}s-{plugin}-query_params"]
 ===== `query_params`
@@ -392,7 +393,6 @@ Accepted options:
 |`named_params` |[] | List of named parameters and their matches used in the `query`
 |===
 
-
 [id="plugins-{type}s-{plugin}-query_template"]
 ===== `query_template` 
 

lib/logstash/filters/elasticsearch.rb

@@ -235,18 +235,6 @@ def prepare_user_agent
 
   private
 
-  # if @target is defined, creates a nested structure to inject result into target field
-  # if not defined, directly sets to the top-level event field
-  # @param event [LogStash::Event]
-  # @param new_key [String] name of the field to set
-  # @param value_to_set [Array] values to set
-  # @return [void]
-  def set_to_event_target(event, new_key, value_to_set)
-    key_to_set = target ? "[#{target}][#{new_key}]" : new_key
-
-    event.set(key_to_set, value_to_set)
-  end
-
   def client_options
     @client_options ||= {
       :user => @user,
@@ -438,9 +426,9 @@ def setup_ssl_params!
     params['ssl_enabled'] = @ssl_enabled ||= Array(@hosts).all? { |host| host && host.to_s.start_with?("https") }
   end
 
-def resolve_query_type
-  @query&.strip&.match?(/\A(?:FROM|ROW|SHOW)/) ? "esql": "dsl"
-end
+  def resolve_query_type
+    @query&.strip&.match?(/\A(?:FROM|ROW|SHOW)/) ? "esql": "dsl"
+  end
 
   def validate_dsl_query_settings!
     #Load query if it exists
@@ -484,17 +472,16 @@ def validate_esql_query_and_params!
     named_params_keys = named_params.map(&:keys).flatten
 
     placeholders = @query.scan(/(?<=[?])[a-z_][a-z0-9_]*/i)
-    raise LogStash::ConfigurationError, "Number of placeholders in `query` and `named_params` do not match" unless placeholders.size == named_params_keys.size
-
     placeholders.each do |placeholder|
       raise LogStash::ConfigurationError, "Placeholder #{placeholder} not found in query" unless named_params_keys.include?(placeholder)
     end
   end
 
   def validate_es_for_esql_support!
     # make sure connected ES supports ES|QL (8.11+)
-    es_supports_esql = Gem::Version.create(get_client.es_version) >= Gem::Version.create(ES_ESQL_SUPPORT_VERSION)
-    fail("Connected Elasticsearch #{get_client.es_version} version does not supports ES|QL. ES|QL feature requires at least Elasticsearch #{ES_ESQL_SUPPORT_VERSION} version.") unless es_supports_esql
+    @es_version ||= get_client.es_version
+    es_supports_esql = Gem::Version.create(@es_version) >= Gem::Version.create(ES_ESQL_SUPPORT_VERSION)
+    fail("Connected Elasticsearch #{@es_version} version does not supports ES|QL. ES|QL feature requires at least Elasticsearch #{ES_ESQL_SUPPORT_VERSION} version.") unless es_supports_esql
   end
 
 end #class LogStash::Filters::Elasticsearch

lib/logstash/filters/elasticsearch/client.rb

@@ -68,7 +68,7 @@ def info
       end
 
       def es_version
-        @es_version ||= info&.dig('version', 'number')
+        info&.dig('version', 'number')
       end
 
       def build_flavor

lib/logstash/filters/elasticsearch/dsl_executor.rb

@@ -5,7 +5,6 @@ module Filters
     class Elasticsearch
       class DslExecutor
         def initialize(plugin, logger)
-          @plugin = plugin
           @index = plugin.params["index"]
           @query = plugin.params["query"]
           @query_dsl = plugin.query_dsl
@@ -17,6 +16,13 @@ def initialize(plugin, logger)
           @sort = plugin.params["sort"]
           @aggregation_fields = plugin.params["aggregation_fields"]
           @logger = logger
+          @event_decorator = plugin.method(:decorate)
+          @target_field = plugin.params["target"]
+          if @target_field
+            def self.apply_target(path); "[#{@target_field}][#{path}]"; end
+          else
+            def self.apply_target(path); path; end
+          end
         end
 
         def process(client, event)
@@ -46,25 +52,27 @@ def process(client, event)
               matched = true
               @fields.each do |old_key, new_key|
                 old_key_path = extract_path(old_key)
-                set = result_hits.map do |doc|
+                extracted_hit_values = result_hits.map do |doc|
                   extract_value(doc["_source"], old_key_path)
                 end
-                event.set(new_key, set.count > 1 ? set : set.first)
+                value_to_set = extracted_hit_values.count > 1 ? extracted_hit_values : extracted_hit_values.first
+                set_to_event_target(event, new_key, value_to_set)
               end
               @docinfo_fields.each do |old_key, new_key|
                 old_key_path = extract_path(old_key)
-                set = result_hits.map do |doc|
+                extracted_docs_info = result_hits.map do |doc|
                   extract_value(doc, old_key_path)
                 end
-                event.set(new_key, set.count > 1 ? set : set.first)
+                value_to_set = extracted_docs_info.count > 1 ? extracted_docs_info : extracted_docs_info.first
+                set_to_event_target(event, new_key, value_to_set)
               end
             end
 
             result_aggregations = results["aggregations"]
             if !result_aggregations.nil? && !result_aggregations.empty?
               matched = true
               @aggregation_fields.each do |agg_name, ls_field|
-                event.set(ls_field, result_aggregations[agg_name])
+                set_to_event_target(event, ls_field, result_aggregations[agg_name])
               end
             end
 
@@ -78,7 +86,7 @@ def process(client, event)
             end
             @tag_on_failure.each { |tag| event.tag(tag) }
           else
-            @plugin.decorate(event) if matched
+            @event_decorator.call(event) if matched
           end
         end
 
@@ -116,6 +124,16 @@ def extract_value(source, path)
           end
         end
 
+        # if @target is defined, creates a nested structure to inject result into target field
+        # if not defined, directly sets to the top-level event field
+        # @param event [LogStash::Event]
+        # @param new_key [String] name of the field to set
+        # @param value_to_set [Array] values to set
+        # @return [void]
+        def set_to_event_target(event, new_key, value_to_set)
+          key_to_set = self.apply_target(new_key)
+          event.set(key_to_set, value_to_set)
+        end
       end
     end
   end

lib/logstash/filters/elasticsearch/esql_executor.rb

@@ -6,9 +6,9 @@ class Elasticsearch
       class EsqlExecutor
 
         def initialize(plugin, logger)
-          @plugin = plugin
           @logger = logger
 
+          @event_decorator = plugin.method(:decorate)
           @query = plugin.params["query"]
           if @query.strip.start_with?("FROM") && !@query.match?(/\|\s*LIMIT/)
             @logger.warn("ES|QL query doesn't contain LIMIT, adding `| LIMIT 1` to optimize the performance")
@@ -20,14 +20,21 @@ def initialize(plugin, logger)
           @fields = plugin.params["fields"]
           @tag_on_failure = plugin.params["tag_on_failure"]
           @logger.debug("ES|QL query executor initialized with ", query: @query, named_params: @named_params)
+
+          @target_field = plugin.params["target"]
+          if @target_field
+            def self.apply_target(path); "[#{@target_field}][#{path}]"; end
+          else
+            def self.apply_target(path); path; end
+          end
         end
 
         def process(client, event)
           resolved_params = @named_params&.any? ? resolve_parameters(event) : []
           response = execute_query(client, resolved_params)
           inform_warning(response)
           process_response(event, response)
-          @plugin.decorate(event)
+          @event_decorator.call(event)
         rescue => e
           @logger.error("Failed to process ES|QL filter", exception: e)
           @tag_on_failure.each { |tag| event.tag(tag) }
@@ -54,7 +61,7 @@ def resolve_parameters(event)
         def execute_query(client, params)
           # debug logs  may help to check what query shape the plugin is sending to ES
           @logger.debug("Executing ES|QL query", query: @query, params: params)
-          client.esql_query({ body: { query: @query, params: params }, format: 'json', drop_null_columns: true }, 'esql')
+          client.esql_query({ body: { query: @query, params: params }, format: 'json', drop_null_columns: true })
         end
 
         def process_response(event, response)
@@ -88,11 +95,22 @@ def add_requested_fields(event, columns, values)
             column_index = columns.find_index { |col| col['name'] == old_key }
             next unless column_index
 
-            row_values = values[column_index]&.compact # remove non-exist field values with compact
-            # TODO: set to the target field once target support is added
-            event.set(new_key, row_values.one? ? row_values.first : row_values) if row_values&.size > 0
+            row_values = values.map { |entry| entry[column_index] }&.compact
+            value_to_set = row_values.count > 1 ? row_values : row_values.first
+            set_to_event_target(event, new_key, value_to_set) unless value_to_set.nil?
           end
         end
+
+        # if @target is defined, creates a nested structure to inject result into target field
+        # if not defined, directly sets to the top-level event field
+        # @param event [LogStash::Event]
+        # @param new_key [String] name of the field to set
+        # @param value_to_set [Array] values to set
+        # @return [void]
+        def set_to_event_target(event, new_key, value_to_set)
+          key_to_set = self.apply_target(new_key)
+          event.set(key_to_set, value_to_set)
+        end
       end
     end
   end