fix: prevent sub-batch 413's from blocking 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.

Author: yaauie

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, :payload_size => 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,21 @@ def bulk(actions)
         stream_writer = body_stream
       end
       bulk_responses = []
+      batch_actions = []
       bulk_actions.each do |action|
         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
+          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
+      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 +140,43 @@ def join_bulk_responses(bulk_responses)
       }
     end
 
-    def bulk_send(body_stream)
+    def bulk_send(body_stream, batch_actions)
+      logger.trace("Sending bulk request to Elasticsearch", :action_count => batch_actions.size, :content_length => body_stream.size)
       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 # Entity Too Large
+        logger.warn("Received HTTP/1.1 status `413 Payload Too Large` sending #{batch_actions.size} actions (#{body_stream.size} bytes)")
+        emulate_batch_error_response(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)