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
20
21
22
23
24
 
# File 'lib/logstash/inputs/elasticsearch/aggregation.rb', line 11

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

  @size = @plugin_params["size"]
  @query = @plugin_params["query"]
  @retries = @plugin_params["retries"]
  @agg_options = {
    :index => @index,
    :size  => 0
  }.merge(:body => @query)

  @plugin = plugin
end

Instance Method Details

#do_run(output_queue) ⇒ Object 



36
37
38
39
40
41
42
 
# File 'lib/logstash/inputs/elasticsearch/aggregation.rb', line 36

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

#retryable(job_name, &block) ⇒ Object 



26
27
28
29
30
31
32
33
34
 
# File 'lib/logstash/inputs/elasticsearch/aggregation.rb', line 26

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