Class: LogStash::Inputs::Elasticsearch

Inherits:
Base
  • Object
show all
Extended by:
PositiveWholeNumberValidator, URIOrEmptyValidator, PluginMixins::ValidatorSupport::FieldReferenceValidationAdapter
Includes:
PluginMixins::CATrustedFingerprintSupport, PluginMixins::ECSCompatibilitySupport::TargetCheck, PluginMixins::EventSupport::EventFactoryAdapter, PluginMixins::NormalizeConfigSupport, PluginMixins::Scheduler
Defined in:
lib/logstash/inputs/elasticsearch.rb,
lib/logstash/inputs/elasticsearch/esql.rb,
lib/logstash/inputs/elasticsearch/aggregation.rb,
lib/logstash/inputs/elasticsearch/cursor_tracker.rb,
lib/logstash/inputs/elasticsearch/paginated_search.rb

Overview

.Compatibility Note [NOTE]

Starting with Elasticsearch 5.3, there's an refmodules-http.html[HTTP setting] called http.content_type.required. If this option is set to true, and you are using Logstash 2.4 through 5.2, you need to update the Elasticsearch input plugin to version 4.0.2 or higher.

================================================================================

Read from an Elasticsearch cluster, based on search query results. This is useful for replaying test logs, reindexing, etc. It also supports periodically scheduling lookup enrichments using a cron syntax (see schedule setting).

Example: [source,ruby] input { # Read all documents from Elasticsearch matching the given query elasticsearch { hosts => "localhost" query => '{ "query": { "match": { "statuscode": 200 } }, "sort": [ "_doc" ] }' } }

This would create an Elasticsearch query with the following format: [source,json] curl 'http://localhost:9200/logstash-*/_search?&scroll=1m&size=1000' -d '{ "query": { "match": { "statuscode": 200 } }, "sort": [ "_doc" ] }'

Scheduling

Input from this plugin can be scheduled to run periodically according to a specific schedule. This scheduling syntax is powered by scheduler">https://github.com/jmettraux/rufus-scheduler. The syntax is cron-like with some extensions specific to Rufus (e.g. timezone support ).

Examples:

|========================================================== | * 5 * 1-3 * | will execute every minute of 5am every day of January through March. | 0 * * * * | will execute on the 0th minute of every hour every day. | 0 6 * * * America/Chicago | will execute at 6:00am (UTC/GMT -5) every day. |==========================================================

Further documentation describing this syntax can be found strings">https://github.com/jmettraux/rufus-scheduler#parsing-cronlines-and-time-strings.

Defined Under Namespace

Modules: PositiveWholeNumberValidator, URIOrEmptyValidator Classes: Aggregation, ColumnSpec, CursorTracker, Esql, PaginatedSearch, Scroll, SearchAfter

Constant Summary collapse

BUILD_FLAVOR_SERVERLESS = 
'serverless'.freeze
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 =

the version started using elasticsearch-ruby v8

"8.17.4"
ES_ESQL_SUPPORT_VERSION = 
"8.11.0"

Instance Attribute Summary collapse

Instance Method Summary collapse

Methods included from URIOrEmptyValidator

validate_value

Methods included from PositiveWholeNumberValidator

validate_value

Constructor Details

#initialize(params = {}) ⇒ Elasticsearch

Returns a new instance of Elasticsearch.



301
302
303
304
305
306
307
 
# File 'lib/logstash/inputs/elasticsearch.rb', line 301

def initialize(params={})
  super(params)

  if docinfo_target.nil?
    @docinfo_target = ecs_select[disabled: '@metadata', v1: '[@metadata][input][elasticsearch]']
  end
end

Instance Attribute Details

#pipeline_id ⇒ Object (readonly)

Returns the value of attribute pipeline_id.



292
293
294
 
# File 'lib/logstash/inputs/elasticsearch.rb', line 292

def pipeline_id
  @pipeline_id
end

Instance Method Details

#decorate_event(event) ⇒ Object 



373
374
375
 
# File 'lib/logstash/inputs/elasticsearch.rb', line 373

def decorate_event(event)
  decorate(event)
end

#push_hit(hit, output_queue, root_field = '_source') ⇒ Object 



366
367
368
369
370
371
 
# File 'lib/logstash/inputs/elasticsearch.rb', line 366

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

#register ⇒ Object 



309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
 
# File 'lib/logstash/inputs/elasticsearch.rb', line 309

def register
  require "rufus/scheduler"

  @pipeline_id = execution_context&.pipeline_id || 'main'

  fill_hosts_from_cloud_id
  setup_ssl_params!

  if @query_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 response_type tracking_field)
    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}`")

  validate_authentication
  fill_user_password_from_cloud_auth

  @client = new_client

  test_connection!

  validate_es_for_esql_support!

  setup_serverless

  setup_search_api

  @query_executor = create_query_executor

  setup_cursor_tracker

  @client
end

#run(output_queue) ⇒ Object 



352
353
354
355
356
357
358
359
360
361
 
# File 'lib/logstash/inputs/elasticsearch.rb', line 352

def run(output_queue)
  if @schedule
    scheduler.cron(@schedule, :overlap => @schedule_overlap) do
      @query_executor.do_run(output_queue, get_query_object())
    end
    scheduler.join
  else
    @query_executor.do_run(output_queue, get_query_object())
  end
end