Class: LogStash::Inputs::Elasticsearch::Aggregation

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

Constant Summary collapse

AGGREGATION_JOB = 
"aggregation"

Instance Method Summary collapse

Constructor Details

#initialize(client, plugin) ⇒ Aggregation

Returns a new instance of Aggregation.



11
12
13
14
15
16
17
18
19
 
# File 'lib/logstash/inputs/elasticsearch/aggregation.rb', line 11

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

  @index = @plugin_params["index"]
  @size = @plugin_params["size"]
  @retries = @plugin_params["retries"]
  @plugin = plugin
end

Instance Method Details

#aggregation_options(query_object) ⇒ Object 



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

def aggregation_options(query_object)
  {
    :index => @index,
    :size => 0,
    :body => query_object
  }
end

#do_run(output_queue, query_object) ⇒ Object 



39
40
41
42
43
44
45
 
# File 'lib/logstash/inputs/elasticsearch/aggregation.rb', line 39

def do_run(output_queue, query_object)
  logger.info("Aggregation starting")
  r = retryable(AGGREGATION_JOB) do
    @client.search(aggregation_options(query_object))
  end
  @plugin.push_hit(r, output_queue, 'aggregations') if r
end

#retryable(job_name, &block) ⇒ Object 



21
22
23
24
25
26
27
28
29
 
# File 'lib/logstash/inputs/elasticsearch/aggregation.rb', line 21

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)
  false
end