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.3
+ - 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)
+
 ## 10.7.2
  - [DOC] Fixed links to restructured Logstash-to-cloud docs [#975](https://github.com/logstash-plugins/logstash-output-elasticsearch/pull/975)
 

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)

logstash-output-elasticsearch.gemspec

@@ -1,6 +1,6 @@
 Gem::Specification.new do |s|
   s.name            = 'logstash-output-elasticsearch'
-  s.version         = '10.7.2'
+  s.version         = '10.7.3'
 
   s.licenses        = ['apache-2.0']
   s.summary         = "Stores logs in Elasticsearch"

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] }