Support aggregations (#202)

Add `response_type` configuration option to allow processing result of aggregations.

The default `hits` will generate one event per returned document (i.e. "hit"), which is the current behavior.

When set to `aggregations`, a single Logstash event will be generated with the contents of the `aggregations` object of the query's response. In this case the `hits` object will be ignored.
The parameter `size` will be always be set to 0 regardless of the default or user-defined value set in this plugin.

---------
Co-authored-by: MikhailMS <[email protected]>

Author: jsvd

CHANGELOG.md

@@ -1,3 +1,6 @@
+## 4.20.0
+  - Added `response_type` configuration option to allow processing result of aggregations [#202](https://github.com/logstash-plugins/logstash-input-elasticsearch/pull/202)
+
 ## 4.19.1
   - Plugin version bump to pick up docs fix in  [#199](https://github.com/logstash-plugins/logstash-input-elasticsearch/pull/199) required to clear build error in docgen. [#200](https://github.com/logstash-plugins/logstash-input-elasticsearch/pull/200)
 

docs/index.asciidoc

@@ -115,6 +115,7 @@ This plugin supports the following configuration options plus the <<plugins-{typ
 | <<plugins-{type}s-{plugin}-password>> |<<password,password>>|No
 | <<plugins-{type}s-{plugin}-proxy>> |<<uri,uri>>|No
 | <<plugins-{type}s-{plugin}-query>> |<<string,string>>|No
+| <<plugins-{type}s-{plugin}-response_type>> |<<string,string>>, one of `["hits","aggregations"]`|No
 | <<plugins-{type}s-{plugin}-request_timeout_seconds>> | <<number,number>>|No
 | <<plugins-{type}s-{plugin}-schedule>> |<<string,string>>|No
 | <<plugins-{type}s-{plugin}-scroll>> |<<string,string>>|No
@@ -337,6 +338,20 @@ documentation] for more information.
 When <<plugins-{type}s-{plugin}-search_api>> resolves to `search_after` and the query does not specify `sort`,
 the default sort `'{ "sort": { "_shard_doc": "asc" } }'` will be added to the query. Please refer to the {ref}/paginate-search-results.html#search-after[Elasticsearch search_after] parameter to know more.
 
+[id="plugins-{type}s-{plugin}-response_type"]
+===== `response_type`
+
+  * Value can be any of: `hits`, `aggregations`
+  * Default value is `hits`
+
+Which part of the result to transform into Logstash events when processing the
+response from the query.
+The default `hits` will generate one event per returned document (i.e. "hit").
+When set to `aggregations`, a single Logstash event will be generated with the
+contents of the `aggregations` object of the query's response. In this case the
+`hits` object will be ignored. The parameter `size` will be always be set to
+0 regardless of the default or user-defined value set in this plugin.
+
 [id="plugins-{type}s-{plugin}-request_timeout_seconds"]
 ===== `request_timeout_seconds`
 

lib/logstash/inputs/elasticsearch.rb

@@ -74,6 +74,7 @@
 class LogStash::Inputs::Elasticsearch < LogStash::Inputs::Base
 
   require 'logstash/inputs/elasticsearch/paginated_search'
+  require 'logstash/inputs/elasticsearch/aggregation'
 
   include LogStash::PluginMixins::ECSCompatibilitySupport(:disabled, :v1, :v8 => :v1)
   include LogStash::PluginMixins::ECSCompatibilitySupport::TargetCheck
@@ -101,6 +102,11 @@ class LogStash::Inputs::Elasticsearch < LogStash::Inputs::Base
   # https://www.elastic.co/guide/en/elasticsearch/reference/current/query-dsl.html
   config :query, :validate => :string, :default => '{ "sort": [ "_doc" ] }'
 
+  # This allows you to speccify the response type: either hits or aggregations
+  # where hits: normal search request
+  #       aggregations: aggregation request
+  config :response_type, :validate => ['hits', 'aggregations'], :default => 'hits'
+
   # This allows you to set the maximum number of hits returned per scroll.
   config :size, :validate => :number, :default => 1000
 
@@ -282,11 +288,6 @@ def register
     fill_hosts_from_cloud_id
     setup_ssl_params!
 
-    @options = {
-      :index => @index,
-      :scroll => @scroll,
-      :size => @size
-    }
     @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")
@@ -328,21 +329,25 @@ def register
 
     setup_search_api
 
+    setup_query_executor
+
     @client
   end
 
-
   def run(output_queue)
     if @schedule
-      scheduler.cron(@schedule) { @paginated_search.do_run(output_queue) }
+      scheduler.cron(@schedule) { @query_executor.do_run(output_queue) }
       scheduler.join
     else
-      @paginated_search.do_run(output_queue)
+      @query_executor.do_run(output_queue)
     end
   end
 
-  def push_hit(hit, output_queue)
-    event = targeted_event_factory.new_event hit['_source']
+  ##
+  # This can be called externally from the query_executor
+  public
+  def push_hit(hit, output_queue, root_field = '_source')
+    event = targeted_event_factory.new_event hit[root_field]
     set_docinfo_fields(hit, event) if @docinfo
     decorate(event)
     output_queue << event
@@ -643,13 +648,20 @@ def setup_search_api
                              @search_api
                            end
 
+  end
 
-    @paginated_search = if @resolved_search_api == "search_after"
+  def setup_query_executor
+    @query_executor = case @response_type
+                      when 'hits'
+                        if @resolved_search_api == "search_after"
                           LogStash::Inputs::Elasticsearch::SearchAfter.new(@client, self)
                         else
                           logger.warn("scroll API is no longer recommended for pagination. Consider using search_after instead.") if es_major_version >= 8
                           LogStash::Inputs::Elasticsearch::Scroll.new(@client, self)
                         end
+                      when 'aggregations'
+                        LogStash::Inputs::Elasticsearch::Aggregation.new(@client, self)
+                      end
   end
 
   module URIOrEmptyValidator

lib/logstash/inputs/elasticsearch/aggregation.rb

@@ -0,0 +1,45 @@
+require 'logstash/helpers/loggable_try'
+
+module LogStash
+  module Inputs
+    class Elasticsearch
+      class Aggregation
+        include LogStash::Util::Loggable
+
+        AGGREGATION_JOB = "aggregation"
+
+        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
+
+        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
+
+        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')
+        end
+      end
+    end
+  end
+end

logstash-input-elasticsearch.gemspec

@@ -1,7 +1,7 @@
 Gem::Specification.new do |s|
 
   s.name            = 'logstash-input-elasticsearch'
-  s.version         = '4.19.1'
+  s.version         = '4.20.0'
   s.licenses        = ['Apache License (2.0)']
   s.summary         = "Reads query results from an Elasticsearch cluster"
   s.description     = "This gem is a Logstash plugin required to be installed on top of the Logstash core pipeline using $LS_HOME/bin/logstash-plugin install gemname. This gem is not a stand-alone program"

spec/inputs/elasticsearch_spec.rb

@@ -112,14 +112,14 @@
       context "ES 8" do
         let(:es_version) { "8.10.0" }
         it "resolves `auto` to `search_after`" do
-          expect(plugin.instance_variable_get(:@paginated_search)).to be_a LogStash::Inputs::Elasticsearch::SearchAfter
+          expect(plugin.instance_variable_get(:@query_executor)).to be_a LogStash::Inputs::Elasticsearch::SearchAfter
         end
       end
 
       context "ES 7" do
         let(:es_version) { "7.17.0" }
         it "resolves `auto` to `scroll`" do
-          expect(plugin.instance_variable_get(:@paginated_search)).to be_a LogStash::Inputs::Elasticsearch::Scroll
+          expect(plugin.instance_variable_get(:@query_executor)).to be_a LogStash::Inputs::Elasticsearch::Scroll
         end
       end
     end
@@ -268,7 +268,7 @@
       before { plugin.register }
 
       it 'runs just one slice' do
-        expect(plugin.instance_variable_get(:@paginated_search)).to receive(:search).with(duck_type(:<<), nil)
+        expect(plugin.instance_variable_get(:@query_executor)).to receive(:search).with(duck_type(:<<), nil)
         expect(Thread).to_not receive(:new)
 
         plugin.run([])
@@ -280,7 +280,7 @@
       before { plugin.register }
 
       it 'runs just one slice' do
-        expect(plugin.instance_variable_get(:@paginated_search)).to receive(:search).with(duck_type(:<<), nil)
+        expect(plugin.instance_variable_get(:@query_executor)).to receive(:search).with(duck_type(:<<), nil)
         expect(Thread).to_not receive(:new)
 
         plugin.run([])
@@ -295,7 +295,7 @@
         it "runs #{slice_count} independent slices" do
           expect(Thread).to receive(:new).and_call_original.exactly(slice_count).times
           slice_count.times do |slice_id|
-            expect(plugin.instance_variable_get(:@paginated_search)).to receive(:search).with(duck_type(:<<), slice_id)
+            expect(plugin.instance_variable_get(:@query_executor)).to receive(:search).with(duck_type(:<<), slice_id)
           end
 
           plugin.run([])
@@ -423,8 +423,8 @@ def synchronize_method!(object, method_name)
           expect(client).to receive(:search).with(hash_including(:body => slice1_query)).and_return(slice1_response0)
           expect(client).to receive(:scroll).with(hash_including(:body => { :scroll_id => slice1_scroll1 })).and_return(slice1_response1)
 
-          synchronize_method!(plugin.instance_variable_get(:@paginated_search), :next_page)
-          synchronize_method!(plugin.instance_variable_get(:@paginated_search), :initial_search)
+          synchronize_method!(plugin.instance_variable_get(:@query_executor), :next_page)
+          synchronize_method!(plugin.instance_variable_get(:@query_executor), :initial_search)
         end
 
         let(:client) { Elasticsearch::Client.new }
@@ -493,14 +493,14 @@ def synchronize_method!(object, method_name)
           expect(client).to receive(:search).with(hash_including(:body => slice1_query)).and_return(slice1_response0)
           expect(client).to receive(:scroll).with(hash_including(:body => { :scroll_id => slice1_scroll1 })).and_raise("boom")
 
-          synchronize_method!(plugin.instance_variable_get(:@paginated_search), :next_page)
-          synchronize_method!(plugin.instance_variable_get(:@paginated_search), :initial_search)
+          synchronize_method!(plugin.instance_variable_get(:@query_executor), :next_page)
+          synchronize_method!(plugin.instance_variable_get(:@query_executor), :initial_search)
         end
 
         let(:client) { Elasticsearch::Client.new }
 
         it 'insert event to queue without waiting other slices' do
-          expect(plugin.instance_variable_get(:@paginated_search)).to receive(:search).twice.and_wrap_original do |m, *args|
+          expect(plugin.instance_variable_get(:@query_executor)).to receive(:search).twice.and_wrap_original do |m, *args|
             q = args[0]
             slice_id = args[1]
             if slice_id == 0
@@ -1020,7 +1020,7 @@ def wait_receive_request
 
     it "should properly schedule" do
       begin
-        expect(plugin.instance_variable_get(:@paginated_search)).to receive(:do_run) {
+        expect(plugin.instance_variable_get(:@query_executor)).to receive(:do_run) {
           queue << LogStash::Event.new({})
         }.at_least(:twice)
         runner = Thread.start { plugin.run(queue) }
@@ -1033,7 +1033,62 @@ def wait_receive_request
         runner.join if runner
       end
     end
+  end
+
+  context "aggregations" do
+    let(:config) do
+      {
+        'hosts'         => ["localhost"],
+        'query'         => '{ "query": {}, "size": 0, "aggs":{"total_count": { "value_count": { "field": "type" }}, "empty_count": { "sum": { "field": "_meta.empty_event" }}}}',
+        'response_type' => 'aggregations',
+        'size'          => 0
+      }
+    end
 
+    let(:mock_response) do
+      {
+        "took" => 27,
+        "timed_out" => false,
+        "_shards" => {
+          "total" => 169,
+          "successful" => 169,
+          "skipped" => 0,
+          "failed" => 0
+        },
+        "hits" => {
+          "total" => 10,
+          "max_score" => 1.0,
+          "hits" => []
+        },
+        "aggregations" => {
+          "total_counter" => {
+            "value" => 10
+          },
+          "empty_counter" => {
+            "value" => 5
+          },
+        }
+      }
+    end
+
+    before(:each) do
+      client = Elasticsearch::Client.new
+
+      expect(Elasticsearch::Client).to receive(:new).with(any_args).and_return(client)
+      expect(client).to receive(:search).with(any_args).and_return(mock_response)
+      expect(client).to receive(:ping)
+    end
+
+    before { plugin.register }
+
+    it 'creates the events from the aggregations' do
+      plugin.run queue
+      event = queue.pop
+
+      expect(event).to be_a(LogStash::Event)
+      expect(event.get("[total_counter][value]")).to eql 10
+      expect(event.get("[empty_counter][value]")).to eql 5
+    end
   end
 
   context "retries" do