Class: LogStash::Inputs::Elasticsearch::PaginatedSearch

Inherits:
Object
  • Object
show all
Includes:
Util::Loggable
Defined in:
lib/logstash/inputs/elasticsearch/paginated_search.rb

Direct Known Subclasses

Scroll, SearchAfter

Instance Method Summary collapse

Constructor Details

#initialize(client, plugin) ⇒ PaginatedSearch

Returns a new instance of PaginatedSearch.



9
10
11
12
13
14
15
16
17
18
19
20
21
22
 
# File 'lib/logstash/inputs/elasticsearch/paginated_search.rb', line 9

def initialize(client, plugin)
  @client = client
  @plugin_params = plugin.params

  @index = @plugin_params["index"]
  @query = LogStash::Json.load(@plugin_params["query"])
  @scroll = @plugin_params["scroll"]
  @size = @plugin_params["size"]
  @slices = @plugin_params["slices"]
  @retries = @plugin_params["retries"]

  @plugin = plugin
  @pipeline_id = plugin.pipeline_id
end

Instance Method Details

#do_run(output_queue) ⇒ Object 



24
25
26
27
28
 
# File 'lib/logstash/inputs/elasticsearch/paginated_search.rb', line 24

def do_run(output_queue)
  return retryable_search(output_queue) if @slices.nil? || @slices <= 1

  retryable_slice_search(output_queue)
end

#retryable(job_name, &block) ⇒ Object 



30
31
32
33
34
35
36
37
 
# File 'lib/logstash/inputs/elasticsearch/paginated_search.rb', line 30

def retryable(job_name, &block)
  stud_try = ::LogStash::Helpers::LoggableTry.new(logger, job_name)
  stud_try.try((@retries + 1).times) { yield }
rescue => e
  error_details = {:message => e.message, :cause => e.cause}
  error_details[:backtrace] = e.backtrace if logger.debug?
  logger.error("Tried #{job_name} unsuccessfully", error_details)
end

#retryable_search(output_queue) ⇒ Object

Raises:

  • (NotImplementedError) 


39
40
41
 
# File 'lib/logstash/inputs/elasticsearch/paginated_search.rb', line 39

def retryable_search(output_queue)
  raise NotImplementedError
end

#retryable_slice_search(output_queue) ⇒ Object

Raises:

  • (NotImplementedError) 


43
44
45
 
# File 'lib/logstash/inputs/elasticsearch/paginated_search.rb', line 43

def retryable_slice_search(output_queue)
  raise NotImplementedError
end