size batch by _decompressed_ payload size

Author: yaauie

CHANGELOG.md

@@ -1,5 +1,7 @@
 ## 10.8.6
  - Fixed an issue where a single over-size event being rejected by Elasticsearch would cause the entire entire batch to be retried indefinitely. The oversize event will still be retried on its own and logging has been improved to include payload sizes in this situation [#972](https://github.com/logstash-plugins/logstash-output-elasticsearch/pull/972)
+ - Fixed an issue with `http_compression => true` where a well-compressed payload could fit under our outbound 20MB limit but expand beyond Elasticsearch's 100MB limit, causing bulk failures. Bulk grouping is now determined entirely by the decompressed payload size [#823](https://github.com/logstash-plugins/logstash-output-elasticsearch/issues/823)
+ - Improved debug-level logging about bulk requests.
 
 ## 10.8.5
  - Feat: assert returned item count from _bulk [#997](https://github.com/logstash-plugins/logstash-output-elasticsearch/pull/997)

lib/logstash/outputs/elasticsearch/http_client.rb

@@ -109,8 +109,8 @@ def bulk(actions)
       body_stream = StringIO.new
       if http_compression
         body_stream.set_encoding "BINARY"
-        stream_writer = Zlib::GzipWriter.new(body_stream, Zlib::DEFAULT_COMPRESSION, Zlib::DEFAULT_STRATEGY)
-      else 
+        stream_writer = gzip_writer(body_stream)
+      else
         stream_writer = body_stream
       end
       bulk_responses = []
@@ -120,21 +120,39 @@ def bulk(actions)
                     action.map {|line| LogStash::Json.dump(line)}.join("\n") :
                     LogStash::Json.dump(action)
         as_json << "\n"
-        if (body_stream.size + as_json.bytesize) > TARGET_BULK_BYTES && body_stream.size > 0
-          logger.debug("Sending partial bulk request for batch with one or more actions remaining.", :action_count => batch_actions.size, :content_length => body_stream.size, :batch_offset => (index + 1 - batch_actions.size))
+        if (stream_writer.pos + as_json.bytesize) > TARGET_BULK_BYTES && stream_writer.pos > 0
+          stream_writer.flush # ensure writer has sync'd buffers before reporting sizes
+          logger.debug("Sending partial bulk request for batch with one or more actions remaining.",
+                       :action_count => batch_actions.size,
+                       :payload_size => stream_writer.pos,
+                       :content_length => body_stream.size,
+                       :batch_offset => (index + 1 - batch_actions.size))
           bulk_responses << bulk_send(body_stream, batch_actions)
+          body_stream.truncate(0) && body_stream.seek(0)
+          stream_writer = gzip_writer(body_stream) if http_compression
           batch_actions.clear
         end
         stream_writer.write(as_json)
         batch_actions << action
       end
       stream_writer.close if http_compression
-      logger.debug("Sending final bulk request for batch.", :action_count => batch_actions.size, :content_length => body_stream.size, :batch_offset => (actions.size - batch_actions.size))
+      logger.debug("Sending final bulk request for batch.",
+                   :action_count => batch_actions.size,
+                   :payload_size => stream_writer.pos,
+                   :content_length => body_stream.size,
+                   :batch_offset => (actions.size - batch_actions.size))
       bulk_responses << bulk_send(body_stream, batch_actions) if body_stream.size > 0
       body_stream.close if !http_compression
       join_bulk_responses(bulk_responses)
     end
 
+    def gzip_writer(io)
+      fail(ArgumentError, "Cannot create gzip writer on IO with unread bytes") unless io.eof?
+      fail(ArgumentError, "Cannot create gzip writer on non-empty IO") unless io.pos == 0
+
+      Zlib::GzipWriter.new(io, Zlib::DEFAULT_COMPRESSION, Zlib::DEFAULT_STRATEGY)
+    end
+
     def join_bulk_responses(bulk_responses)
       {
         "errors" => bulk_responses.any? {|r| r["errors"] == true},
@@ -160,11 +178,6 @@ def bulk_send(body_stream, batch_actions)
           response.code, url, body_stream.to_s, response.body
         )
       end
-    ensure
-      if !body_stream.closed?
-        body_stream.truncate(0)
-        body_stream.seek(0)
-      end
     end
 
     def emulate_batch_error_response(actions, http_code, reason)

spec/unit/outputs/elasticsearch/http_client_spec.rb

@@ -204,50 +204,69 @@
   end
 
   describe "#bulk" do
-    subject { described_class.new(base_options) }
+    subject(:http_client) { described_class.new(base_options) }
 
     require "json"
     let(:message) { "hey" }
     let(:actions) { [
       ["index", {:_id=>nil, :_index=>"logstash"}, {"message"=> message}],
     ]}
 
-    context "if a message is over TARGET_BULK_BYTES" do
-      let(:target_bulk_bytes) { LogStash::Outputs::ElasticSearch::TARGET_BULK_BYTES }
-      let(:message) { "a" * (target_bulk_bytes + 1) }
+    [true,false].each do |http_compression_enabled|
+      context "with `http_compression => #{http_compression_enabled}`" do
 
-      it "should be handled properly" do
-        allow(subject).to receive(:join_bulk_responses)
-        expect(subject).to receive(:bulk_send).once do |data|
-          expect(data.size).to be > target_bulk_bytes
+        let(:base_options) { super().merge(:client_settings => {:http_compression => http_compression_enabled}) }
+
+        before(:each) do
+          if http_compression_enabled
+            expect(http_client).to receive(:gzip_writer).at_least(:once).and_call_original
+          else
+            expect(http_client).to_not receive(:gzip_writer)
+          end
         end
-        s = subject.send(:bulk, actions)
-      end
-    end
 
-    context "with two messages" do
-      let(:message1) { "hey" }
-      let(:message2) { "you" }
-      let(:actions) { [
-        ["index", {:_id=>nil, :_index=>"logstash"}, {"message"=> message1}],
-        ["index", {:_id=>nil, :_index=>"logstash"}, {"message"=> message2}],
-      ]}
-      it "executes one bulk_send operation" do
-        allow(subject).to receive(:join_bulk_responses)
-        expect(subject).to receive(:bulk_send).once
-        s = subject.send(:bulk, actions)
-      end
+        context "if a message is over TARGET_BULK_BYTES" do
+          let(:target_bulk_bytes) { LogStash::Outputs::ElasticSearch::TARGET_BULK_BYTES }
+          let(:message) { "a" * (target_bulk_bytes + 1) }
+
+          it "should be handled properly" do
+            allow(subject).to receive(:join_bulk_responses)
+            expect(subject).to receive(:bulk_send).once do |data|
+              if !http_compression_enabled
+                expect(data.size).to be > target_bulk_bytes
+              else
+                expect(Zlib::gunzip(data.string).size).to be > target_bulk_bytes
+              end
+            end
+            s = subject.send(:bulk, actions)
+          end
+        end
+
+        context "with two messages" do
+          let(:message1) { "hey" }
+          let(:message2) { "you" }
+          let(:actions) { [
+            ["index", {:_id=>nil, :_index=>"logstash"}, {"message"=> message1}],
+            ["index", {:_id=>nil, :_index=>"logstash"}, {"message"=> message2}],
+          ]}
+          it "executes one bulk_send operation" do
+            allow(subject).to receive(:join_bulk_responses)
+            expect(subject).to receive(:bulk_send).once
+            s = subject.send(:bulk, actions)
+          end
 
-      context "if one exceeds TARGET_BULK_BYTES" do
-        let(:target_bulk_bytes) { LogStash::Outputs::ElasticSearch::TARGET_BULK_BYTES }
-        let(:message1) { "a" * (target_bulk_bytes + 1) }
-        it "executes two bulk_send operations" do
-          allow(subject).to receive(:join_bulk_responses)
-          expect(subject).to receive(:bulk_send).twice
-          s = subject.send(:bulk, actions)
+          context "if one exceeds TARGET_BULK_BYTES" do
+            let(:target_bulk_bytes) { LogStash::Outputs::ElasticSearch::TARGET_BULK_BYTES }
+            let(:message1) { "a" * (target_bulk_bytes + 1) }
+            it "executes two bulk_send operations" do
+              allow(subject).to receive(:join_bulk_responses)
+              expect(subject).to receive(:bulk_send).twice
+              s = subject.send(:bulk, actions)
+            end
+          end
         end
-      end
-    end
+       end
+     end
   end
 
   describe "sniffing" do

spec/unit/outputs/elasticsearch_spec.rb

@@ -328,7 +328,7 @@
     end
 
       before(:each) do
-        allow(subject.client).to receive(:bulk_send).with(instance_of(StringIO)) do |stream|
+        allow(subject.client).to receive(:bulk_send).with(instance_of(StringIO), instance_of(Array)) do |stream, actions|
           expect( stream.string ).to include '"foo":"bar1"'
           expect( stream.string ).to include '"foo":"bar2"'
         end.and_return(bulk_response, {"errors"=>false}) # let's make it go away (second call) to not retry indefinitely