Add filter_path to bulk messages (#1154)

This commit sets the `filter_path` query parameter when sending messages
to Elasticsearch using the bulk API. This should significantly reduce
the size of the query response from Elasticsearch, which should help
reduce bandwidth usage, and improve response processing speed due to
the lesser amount of JSON to deserialize

* Add query to expected results to fix integration tests
* Remork PR to respect exiting `filter_path` settings in custom bulk endpoints
* Add doc and comment explaining filter_path addition to /_bulk

Resolves: #1153

Author: robbavey

CHANGELOG.md

@@ -1,3 +1,6 @@
+## 11.19.0
+  - Added `filter_path` to bulk requests to reduce the size of responses from elasticsearch
+  
 ## 11.18.0
   - Added request header `Elastic-Api-Version` for serverless [#1147](https://github.com/logstash-plugins/logstash-output-elasticsearch/pull/1147)
 

docs/index.asciidoc

@@ -423,10 +423,12 @@ Elasticsearch {ref}/security-api-create-api-key.html[Create API key API].
 ===== `bulk_path`
 
   * Value type is <<string,string>>
-  * There is no default value for this setting.
+  * The default value for this settings is `/_bulk?filter_path=errors,items.*.error,items.*.status`
 
 HTTP Path to perform the _bulk requests to
-this defaults to a concatenation of the path parameter and "_bulk"
+* This default bulk path is the concatenation of the value of `path` parameter and `/_bulk?filter_path=errors,items.*.error,items.*.status`
+* The `filter_path` query parameter is appended to the bulk path to reduce the payload between logstash and elasticsearch. However, if a custom `filter_path` query parameter is included in the `bulk_path` setting, then that value will be used.
+
 
 [id="plugins-{type}s-{plugin}-ca_trusted_fingerprint"]
 ===== `ca_trusted_fingerprint`

lib/logstash/outputs/elasticsearch/http_client.rb

@@ -177,6 +177,7 @@ def join_bulk_responses(bulk_responses)
 
     def bulk_send(body_stream, batch_actions)
       params = compression_level? ? {:headers => {"Content-Encoding" => "gzip"}} : {}
+
       response = @pool.post(@bulk_path, params, body_stream.string)
 
       @bulk_response_metrics.increment(response.code.to_s)

lib/logstash/outputs/elasticsearch/http_client_builder.rb

@@ -33,9 +33,9 @@ def self.build(logger, hosts, params)
       end
 
       common_options[:bulk_path] = if params["bulk_path"]
-         dedup_slashes("/#{params["bulk_path"]}")
+        resolve_filter_path(dedup_slashes("/#{params["bulk_path"]}"))
       else
-         dedup_slashes("/#{params["path"]}/_bulk")
+        resolve_filter_path(dedup_slashes("/#{params["path"]}/_bulk"))
       end
 
       common_options[:sniffing_path] = if params["sniffing_path"]
@@ -197,5 +197,16 @@ def self.setup_api_key(logger, params)
     def self.dedup_slashes(url)
       url.gsub(/\/+/, "/")
     end
+
+    # Set a `filter_path` query parameter if it is not already set to be
+    # `filter_path=errors,items.*.error,items.*.status` to reduce the payload between Logstash and Elasticsearch
+    def self.resolve_filter_path(url)
+      return url if url.match?(/(?:[&|?])filter_path=/)
+      ("#{url}#{query_param_separator(url)}filter_path=errors,items.*.error,items.*.status")
+    end
+
+    def self.query_param_separator(url)
+      url.match?(/\?[^\s#]+/) ? '&' : '?'
+    end
   end
 end; end; end

logstash-output-elasticsearch.gemspec

@@ -1,6 +1,6 @@
 Gem::Specification.new do |s|
   s.name            = 'logstash-output-elasticsearch'
-  s.version         = '11.18.0'
+  s.version         = '11.19.0'
   s.licenses        = ['apache-2.0']
   s.summary         = "Stores logs in Elasticsearch"
   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/integration/outputs/index_spec.rb

@@ -156,7 +156,7 @@ def curl_and_get_json_response(url, method: :get, retrieve_err_payload: false);
   let(:config) { "not implemented" }
   let(:events) { event_count.times.map { event }.to_a }
   subject { LogStash::Outputs::ElasticSearch.new(config) }
-  
+  let(:filter_path) { "filter_path=errors,items.*.error,items.*.status"}
   let(:es_url) { "http://#{get_host_port}" }
   let(:index_url) { "#{es_url}/#{index}" }
 
@@ -178,7 +178,7 @@ def curl_and_get_json_response(url, method: :get, retrieve_err_payload: false);
     subject.do_close
   end
 
-  shared_examples "an indexer" do |secure|
+  shared_examples "an indexer" do |secure, expected_path|
     before(:each) do
       host_unreachable_error_class = LogStash::Outputs::ElasticSearch::HttpClient::Pool::HostUnreachableError
       allow(host_unreachable_error_class).to receive(:new).with(any_args).and_wrap_original do |m, original, url|
@@ -212,13 +212,13 @@ def curl_and_get_json_response(url, method: :get, retrieve_err_payload: false);
         expect(doc["_index"]).to eq(index)
       end
     end
-    
+
     it "sets the correct content-type header" do
       expected_manticore_opts = {:headers => {"Content-Type" => "application/json"}, :body => anything}
       if secure
         expected_manticore_opts = {
-          :headers => {"Content-Type" => "application/json"}, 
-          :body => anything, 
+          :headers => {"Content-Type" => "application/json"},
+          :body => anything,
           :auth => {
             :user => user,
             :password => password,
@@ -230,6 +230,20 @@ def curl_and_get_json_response(url, method: :get, retrieve_err_payload: false);
         and_call_original
       subject.multi_receive(events)
     end
+
+    it "sets the bulk path URL and filter path parameter correctly" do
+      expect(subject.client.pool.adapter.client).to receive(:send).
+          with(anything, expected_path != nil ? expected_path : anything, anything).at_least(:once).and_call_original
+      subject.multi_receive(events)
+    end
+
+    it "receives a filtered response" do
+      expect(subject.client).to receive(:join_bulk_responses).
+          with([{"errors"=>false, "items"=>[{"index"=>{"status"=>201}}]}]).
+          and_call_original
+      subject.multi_receive([event])
+    end
+
   end
 
   shared_examples "PKIX path failure" do
@@ -269,6 +283,65 @@ def curl_and_get_json_response(url, method: :get, retrieve_err_payload: false);
     it_behaves_like("an indexer")
   end
 
+  describe "an indexer with custom bulk path", :integration => true do
+    let(:bulk_path) { "/_bulk?routing=true"}
+    let(:config) {
+      {
+          "hosts" => get_host_port,
+          "index" => index,
+          "http_compression" => false,
+          "bulk_path" => bulk_path
+      }
+    }
+    it_behaves_like("an indexer", false) do
+      let (:expected_path) { "#{es_url}#{bulk_path}&#{filter_path}" }
+    end
+  end
+
+  describe "an indexer with filter path as second parameter", :integration => true do
+    let(:bulk_path) { "/_bulk?routing=true&#{filter_path}"}
+    let(:config) {
+      {
+          "hosts" => get_host_port,
+          "index" => index,
+          "http_compression" => false,
+          "bulk_path" => bulk_path
+      }
+    }
+    it_behaves_like("an indexer", false) do
+      let (:expected_path) { "#{es_url}/#{bulk_path}" }
+    end
+  end
+
+  describe "an indexer with filter path as first  parameter", :integration => true do
+    let(:bulk_path) { "/_bulk?#{filter_path}&routing=true"}
+    let(:config) {
+      {
+          "hosts" => get_host_port,
+          "index" => index,
+          "http_compression" => false,
+          "bulk_path" => bulk_path
+      }
+    }
+    it_behaves_like("an indexer", false) do
+      let (:expected_path) { "#{es_url}/#{bulk_path}" }
+    end
+  end
+
+  describe "an indexer with the standard bulk path", :integration => true do
+    let(:config) {
+      {
+          "hosts" => get_host_port,
+          "index" => index,
+          "http_compression" => false
+      }
+    }
+    it_behaves_like("an indexer", false) do
+      let (:expected_path) { "#{es_url}/_bulk?#{filter_path}" }
+    end
+
+  end
+
   describe "an indexer with no type value set (default to doc)", :integration => true do
     let(:type) { ESHelper.es_version_satisfies?("< 7") ? "doc" : "_doc" }
     let(:config) {
@@ -296,7 +369,7 @@ def curl_and_get_json_response(url, method: :get, retrieve_err_payload: false);
         "index" => index,
         "http_compression" => false
       }
-    end 
+    end
 
     let(:curl_opts) { "-u #{user}:#{password}" }
 

spec/unit/http_client_builder_spec.rb

@@ -36,55 +36,70 @@
       end
     end
 
-    describe "healthcheck_path" do
+    describe "bulk_path" do
+      let (:filter_path) {"filter_path=errors,items.*.error,items.*.status"}
+
+      shared_examples("filter_path added to bulk path appropriately") do
+        it "sets the bulk_path option to the expected bulk path" do
+          expect(described_class).to receive(:create_http_client) do |options|
+            expect(options[:bulk_path]).to eq(expected_bulk_path)
+          end
+          described_class.build(logger, hosts, options)
+          end
+      end
 
       context "when setting bulk_path" do
         let(:bulk_path) { "/meh" }
         let(:options) { super().merge("bulk_path" => bulk_path) }
 
         context "when using path" do
           let(:options) { super().merge("path" => "/path") }
-          it "ignores the path setting" do
-            expect(described_class).to receive(:create_http_client) do |options|
-              expect(options[:bulk_path]).to eq(bulk_path)
-            end
-            described_class.build(logger, hosts, options)
-          end
+          let(:expected_bulk_path) { "#{bulk_path}?#{filter_path}" }
+
+          it_behaves_like "filter_path added to bulk path appropriately"
+        end
+
+        context "when setting a filter path as first parameter"  do
+          let (:filter_path) {"filter_path=error"}
+          let(:bulk_path) { "/meh?#{filter_path}&routing=true" }
+          let(:expected_bulk_path) { bulk_path }
+
+          it_behaves_like "filter_path added to bulk path appropriately"
+        end
+
+        context "when setting a filter path as second parameter" do
+          let (:filter_path) {"filter_path=error"}
+          let(:bulk_path) { "/meh?routing=true&#{filter_path}" }
+          let(:expected_bulk_path) { bulk_path }
+
+          it_behaves_like "filter_path added to bulk path appropriately"
         end
+
         context "when not using path" do
+          let(:expected_bulk_path) { "#{bulk_path}?#{filter_path}"}
 
-          it "uses the bulk_path setting" do
-            expect(described_class).to receive(:create_http_client) do |options|
-              expect(options[:bulk_path]).to eq(bulk_path)
-            end
-            described_class.build(logger, hosts, options)
-          end
+          it_behaves_like "filter_path added to bulk path appropriately"
         end
       end
 
       context "when not setting bulk_path" do
 
         context "when using path" do
           let(:path) { "/meh" }
+          let(:expected_bulk_path) { "#{path}/_bulk?#{filter_path}"}
           let(:options) { super().merge("path" => path) }
-          it "sets bulk_path to path+_bulk" do
-            expect(described_class).to receive(:create_http_client) do |options|
-              expect(options[:bulk_path]).to eq("#{path}/_bulk")
-            end
-            described_class.build(logger, hosts, options)
-          end
+
+          it_behaves_like "filter_path added to bulk path appropriately"
         end
 
         context "when not using path" do
-          it "sets the bulk_path to _bulk" do
-            expect(described_class).to receive(:create_http_client) do |options|
-              expect(options[:bulk_path]).to eq("/_bulk")
-            end
-            described_class.build(logger, hosts, options)
-          end
+          let(:expected_bulk_path) { "/_bulk?#{filter_path}"}
+
+          it_behaves_like "filter_path added to bulk path appropriately"
         end
       end
     end
+
     describe "healthcheck_path" do
       context "when setting healthcheck_path" do
         let(:healthcheck_path) { "/meh" }