ES|QL support #233
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change | ||||||
|---|---|---|---|---|---|---|---|---|
|
|
@@ -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 | ||||||||
|
|
@@ -96,15 +97,18 @@ class LogStash::Inputs::Elasticsearch < LogStash::Inputs::Base | |||||||
| # The index or alias to search. | ||||||||
| config :index, :validate => :string, :default => "logstash-*" | ||||||||
|
|
||||||||
| # The query to be executed. DSL or ES|QL (when `response_type => 'esql'`) query shape is accepted. | ||||||||
| # Read the following documentations for more info | ||||||||
| # Query 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: one of [hits, aggregations, esql] | ||||||||
| # where | ||||||||
| # hits: normal search request | ||||||||
| # aggregations: aggregation request | ||||||||
| # esql: ES|QL request | ||||||||
| config :response_type, :validate => %w[hits aggregations esql], :default => 'hits' | ||||||||
|
There was a problem hiding this comment. Migrating to
Suggested change
def register
+ @query_type = normalize_config("query_type") do |normalizer|
+ normalizer.with_deprecated_alias("response_type")
+ end || (@query.start_with?('{') ? 'hits' : 'esql')There was a problem hiding this comment. I was thinking to add the deprecation right after this ES|QL change. There was a problem hiding this comment.
If someone starts using this feature, I would rather that their never-possible-before configuration feels "stable" and doesn't require them to go back and deal with deprecation warnings for things that we knew about before shipping the feature.
This is a very good point. The current So: what if we were to keep This would mean:
There was a problem hiding this comment. Introducing However, considering the behavior and user experience, I do also strongly support this (introducing FYI: current CI snapshot unit test steps are broken (CIs with release versions are fine) due to core |
||||||||
|
|
||||||||
| # This allows you to set the maximum number of hits returned per scroll. | ||||||||
| config :size, :validate => :number, :default => 1000 | ||||||||
|
|
@@ -286,6 +290,9 @@ class LogStash::Inputs::Elasticsearch < LogStash::Inputs::Base | |||||||
| DEFAULT_EAV_HEADER = { "Elastic-Api-Version" => "2023-10-31" }.freeze | ||||||||
| INTERNAL_ORIGIN_HEADER = { 'x-elastic-product-origin' => 'logstash-input-elasticsearch'}.freeze | ||||||||
|
|
||||||||
| LS_ESQL_SUPPORT_VERSION = "8.17.4" # the version started using elasticsearch-ruby v8 | ||||||||
| ES_ESQL_SUPPORT_VERSION = "8.11.0" | ||||||||
|
|
||||||||
| def initialize(params={}) | ||||||||
| super(params) | ||||||||
|
|
||||||||
|
|
@@ -302,10 +309,17 @@ def register | |||||||
| fill_hosts_from_cloud_id | ||||||||
| setup_ssl_params! | ||||||||
|
|
||||||||
| if @response_type == 'esql' | ||||||||
| validate_ls_version_for_esql_support! | ||||||||
| validate_esql_query! | ||||||||
| not_allowed_options = original_params.keys & %w(index size slices search_api, docinfo, docinfo_target, docinfo_fields) | ||||||||
| raise(LogStash::ConfigurationError, "Configured #{not_allowed_options} params are not allowed while using ES|QL query") if not_allowed_options&.size > 1 | ||||||||
| else | ||||||||
| @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") | ||||||||
| @slices < 1 && fail(LogStash::ConfigurationError, "Elasticsearch Input Plugin's `slices` option must be greater than zero, got `#{@slices}`") | ||||||||
| end | ||||||||
| end | ||||||||
|
|
||||||||
| @retries < 0 && fail(LogStash::ConfigurationError, "Elasticsearch Input Plugin's `retries` option must be equal or greater than zero, got `#{@retries}`") | ||||||||
|
|
@@ -341,6 +355,8 @@ def register | |||||||
|
|
||||||||
| test_connection! | ||||||||
|
|
||||||||
| validate_es_for_esql_support! | ||||||||
|
|
||||||||
| setup_serverless | ||||||||
|
|
||||||||
| setup_search_api | ||||||||
|
|
@@ -363,16 +379,6 @@ def run(output_queue) | |||||||
| end | ||||||||
| end | ||||||||
|
|
||||||||
| ## | ||||||||
| # This can be called externally from the query_executor | ||||||||
| public | ||||||||
|
|
@@ -383,6 +389,23 @@ def push_hit(hit, output_queue, root_field = '_source') | |||||||
| record_last_value(event) | ||||||||
| end | ||||||||
|
|
||||||||
| def decorate_event(event) | ||||||||
| decorate(event) | ||||||||
| end | ||||||||
|
|
||||||||
| private | ||||||||
|
|
||||||||
| def get_query_object | ||||||||
|
There was a problem hiding this comment. review note: moved to private area |
||||||||
| return @query if @response_type == 'esql' | ||||||||
| if @cursor_tracker | ||||||||
| query = @cursor_tracker.inject_cursor(@query) | ||||||||
| @logger.debug("new query is #{query}") | ||||||||
| else | ||||||||
| query = @query | ||||||||
| end | ||||||||
| LogStash::Json.load(query) | ||||||||
| end | ||||||||
|
|
||||||||
| def record_last_value(event) | ||||||||
| @cursor_tracker.record_last_value(event) if @tracking_field | ||||||||
| end | ||||||||
|
|
@@ -414,8 +437,6 @@ def set_docinfo_fields(hit, event) | |||||||
| event.set(@docinfo_target, docinfo_target) | ||||||||
| end | ||||||||
|
|
||||||||
| def hosts_default?(hosts) | ||||||||
| hosts.nil? || ( hosts.is_a?(Array) && hosts.empty? ) | ||||||||
| end | ||||||||
|
|
@@ -675,6 +696,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 | ||||||||
|
|
||||||||
|
|
@@ -714,6 +737,26 @@ def get_transport_client_class | |||||||
| ::Elastic::Transport::Transport::HTTP::Manticore | ||||||||
| end | ||||||||
|
|
||||||||
| def validate_ls_version_for_esql_support! | ||||||||
| if Gem::Version.create(LOGSTASH_VERSION) < Gem::Version.create(LS_ESQL_SUPPORT_VERSION) | ||||||||
| fail("Current version of Logstash does not include Elasticsearch client which supports ES|QL. Please upgrade Logstash to at least #{LS_ESQL_SUPPORT_VERSION}") | ||||||||
| 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 validate_es_for_esql_support! | ||||||||
| return unless @response_type == 'esql' | ||||||||
| # make sure connected ES supports ES|QL (8.11+) | ||||||||
| 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 | ||||||||
|
|
||||||||
| module URIOrEmptyValidator | ||||||||
| ## | ||||||||
| # @override to provide :uri_or_empty validator | ||||||||
|
|
||||||||
Uh oh!
There was an error while loading. Please reload this page.