fix: prevent sub-batch 413's from infinitely retrying whole batch

The Http Client breaks batches of actions into sub-batches that are up to 20MB
in size, sending larger actions as batches-of-one, and zips the responses
together to emulate a single batch response from the Elasticsearch API.

When an individual HTTP request is rejected by Elasticsearch (or by an
intermediate proxy or load-balancer) with an HTTP/1.1 413, we can emulate
the error response instead of blowing an exception through to the whole batch.
This allows only the offending events/actions to be subject to retry logic.

Along the way, we improve logging at the `debug` level for sub-batches, and
emit clear `warn`-level logs with payload sizes when we hit HTTP 413 rejections.

Author: yaauie

CHANGELOG.md

@@ -1,3 +1,6 @@
+## 10.7.2
+ - bugfix: prevent a single over-size event from causing entire batch to be retried indefinitely, and improved logging to include payload sizes in this situation [#972](https://github.com/logstash-plugins/logstash-output-elasticsearch/pull/972)
+
 ## 10.7.1
  - [DOC] Document the permissions required in secured clusters [#969](https://github.com/logstash-plugins/logstash-output-elasticsearch/pull/969)
 

lib/logstash/outputs/elasticsearch/common.rb

@@ -414,7 +414,7 @@ def safe_bulk(actions)
         retry unless @stopping.true?
       rescue ::LogStash::Outputs::ElasticSearch::HttpClient::Pool::BadResponseCodeError => e
         @bulk_request_metrics.increment(:failures)
-        log_hash = {:code => e.response_code, :url => e.url.sanitized.to_s}
+        log_hash = {:code => e.response_code, :url => e.url.sanitized.to_s, :content_length => e.request_body.bytesize}
         log_hash[:body] = e.response_body if @logger.debug? # Generally this is too verbose
         message = "Encountered a retryable error. Will Retry with exponential backoff "
 

lib/logstash/outputs/elasticsearch/http_client.rb

@@ -114,18 +114,23 @@ def bulk(actions)
         stream_writer = body_stream
       end
       bulk_responses = []
-      bulk_actions.each do |action|
+      batch_actions = []
+      bulk_actions.each_with_index do |action, index|
         as_json = action.is_a?(Array) ?
                     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
-          bulk_responses << bulk_send(body_stream) unless body_stream.size == 0
+        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))
+          bulk_responses << bulk_send(body_stream, batch_actions)
+          batch_actions.clear
         end
         stream_writer.write(as_json)
+        batch_actions << action
       end
       stream_writer.close if http_compression
-      bulk_responses << bulk_send(body_stream) if body_stream.size > 0
+      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))
+      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
@@ -137,25 +142,42 @@ def join_bulk_responses(bulk_responses)
       }
     end
 
-    def bulk_send(body_stream)
+    def bulk_send(body_stream, batch_actions)
       params = http_compression ? {:headers => {"Content-Encoding" => "gzip"}} : {}
-      # Discard the URL
       response = @pool.post(@bulk_path, params, body_stream.string)
-      if !body_stream.closed?
-        body_stream.truncate(0)
-        body_stream.seek(0)
-      end
 
       @bulk_response_metrics.increment(response.code.to_s)
 
-      if response.code != 200
+      case response.code
+      when 200 # OK
+        LogStash::Json.load(response.body)
+      when 413 # Payload Too Large
+        logger.warn("Bulk request rejected: `413 Payload Too Large`", :action_count => batch_actions.size, :content_length => body_stream.size)
+        emulate_batch_error_response(batch_actions, response.code, 'payload_too_large')
+      else
         url = ::LogStash::Util::SafeURI.new(response.final_url)
         raise ::LogStash::Outputs::ElasticSearch::HttpClient::Pool::BadResponseCodeError.new(
           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
 
-      LogStash::Json.load(response.body)
+    def emulate_batch_error_response(actions, http_code, reason)
+      {
+          "errors" => true,
+          "items" => actions.map do |action|
+            action = action.first if action.is_a?(Array)
+            request_action, request_parameters = action.first
+            {
+                request_action => {"status" => http_code, "error" => { "type" => reason }}
+            }
+          end
+      }
     end
 
     def get(path)

spec/unit/outputs/elasticsearch_spec.rb

@@ -4,7 +4,7 @@
 require "logstash/outputs/elasticsearch"
 
 describe LogStash::Outputs::ElasticSearch do
-  subject { described_class.new(options) }
+  subject(:elasticsearch_output_instance) { described_class.new(options) }
   let(:options) { {} }
   let(:maximum_seen_major_version) { [1,2,5,6,7,8].sample }
 
@@ -265,12 +265,14 @@
       let(:event) { ::LogStash::Event.new("foo" => "bar") }
       let(:error) do
         ::LogStash::Outputs::ElasticSearch::HttpClient::Pool::BadResponseCodeError.new(
-          429, double("url").as_null_object, double("request body"), double("response body")
+          429, double("url").as_null_object, request_body, double("response body")
         )
       end
       let(:logger) { double("logger").as_null_object }
       let(:response) { { :errors => [], :items => [] } }
 
+      let(:request_body) { double(:request_body, :bytesize => 1023) }
+
       before(:each) do
 
         i = 0
@@ -298,6 +300,43 @@
     end
   end
 
+  context '413 errors' do
+    let(:payload_size) { LogStash::Outputs::ElasticSearch::TARGET_BULK_BYTES + 1024 }
+    let(:event) { ::LogStash::Event.new("message" => ("a" * payload_size ) ) }
+
+    let(:logger_stub) { double("logger").as_null_object }
+
+    before(:each) do
+      allow(elasticsearch_output_instance.client).to receive(:logger).and_return(logger_stub)
+
+      allow(elasticsearch_output_instance.client).to receive(:bulk).and_call_original
+
+      max_bytes = payload_size * 3 / 4 # ensure a failure first attempt
+      allow(elasticsearch_output_instance.client.pool).to receive(:post) do |path, params, body|
+        if body.length > max_bytes
+          max_bytes *= 2 # ensure a successful retry
+          double("Response", :code => 413, :body => "")
+        else
+          double("Response", :code => 200, :body => '{"errors":false,"items":[{"index":{"status":200,"result":"created"}}]}')
+        end
+      end
+    end
+
+    it 'retries the 413 until it goes away' do
+      elasticsearch_output_instance.multi_receive([event])
+
+      expect(elasticsearch_output_instance.client).to have_received(:bulk).twice
+    end
+
+    it 'logs about payload quantity and size' do
+      elasticsearch_output_instance.multi_receive([event])
+
+      expect(logger_stub).to have_received(:warn)
+                                 .with(a_string_matching(/413 Payload Too Large/),
+                                       hash_including(:action_count => 1, :content_length => a_value > 20_000_000))
+    end
+  end
+
   context "with timeout set" do
     let(:listener) { Flores::Random.tcp_listener }
     let(:port) { listener[2] }