Make default target_bulk_bytes below AWS limit, but make it configurable

Signed-off-by: Mårten Svantesson <[email protected]>

Author: msvticket

lib/logstash/outputs/opensearch/http_client.rb

@@ -15,23 +15,9 @@
 require 'stringio'
 
 module LogStash; module Outputs; class OpenSearch;
-  # This is a constant instead of a config option because
-  # there really isn't a good reason to configure it.
-  #
-  # The criteria used are:
-  # 1. We need a number that's less than 100MiB because OpenSearch
-  #    won't accept bulks larger than that.
-  # 2. It must be large enough to amortize the connection constant
-  #    across multiple requests.
-  # 3. It must be small enough that even if multiple threads hit this size
-  #    we won't use a lot of heap.
-  #
-  # We wound up agreeing that a number greater than 10 MiB and less than 100MiB
-  # made sense. We picked one on the lowish side to not use too much heap.
-  TARGET_BULK_BYTES = 20 * 1024 * 1024 # 20MiB
-
   class HttpClient
-    attr_reader :client, :options, :logger, :pool, :action_count, :recv_count
+    attr_reader :client, :options, :logger, :pool, :action_count, :recv_count, :target_bulk_bytes
+
     # This is here in case we use DEFAULT_OPTIONS in the future
     # DEFAULT_OPTIONS = {
     #   :setting => value
@@ -73,6 +59,8 @@ def initialize(options={})
       # mutex to prevent requests and sniffing to access the
       # connection pool at the same time
       @bulk_path = @options[:bulk_path]
+
+      @target_bulk_bytes = @options[:target_bulk_bytes]
     end
 
     def build_url_template
@@ -105,7 +93,6 @@ def maximum_seen_major_version
     def bulk(actions)
       @action_count ||= 0
       @action_count += actions.size
-
       return if actions.empty?
 
       bulk_actions = actions.collect do |action, args, source|
@@ -132,7 +119,7 @@ def bulk(actions)
                     action.map {|line| LogStash::Json.dump(line)}.join("\n") :
                     LogStash::Json.dump(action)
         as_json << "\n"
-        if (stream_writer.pos + as_json.bytesize) > TARGET_BULK_BYTES && stream_writer.pos > 0
+        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,

lib/logstash/outputs/opensearch/http_client_builder.rb

@@ -35,6 +35,7 @@ def self.build(logger, hosts, params)
       end
 
       common_options[:timeout] = params["timeout"] if params["timeout"]
+      common_options[:target_bulk_bytes] = params["target_bulk_bytes"]
 
       if params["path"]
         client_settings[:path] = dedup_slashes("/#{params["path"]}/")

lib/logstash/plugin_mixins/opensearch/api_configs.rb

@@ -46,6 +46,20 @@ module APIConfigs
         # this defaults to a concatenation of the path parameter and "_bulk"
         :bulk_path => { :validate => :string },
 
+        # Maximum number of bytes in bulk requests
+        # The criteria for deciding the default value of target_bulk_bytes is:
+        # 1. We need a number that's less than 10MiB because OpenSearch is commonly
+        #    configured (particular in AWS Opensearch Service) to not accept
+        #    bulks larger than that.
+        # 2. It must be large enough to amortize the connection constant
+        #    across multiple requests.
+        # 3. It must be small enough that even if multiple threads hit this size
+        #    we won't use a lot of heap.
+        :target_bulk_bytes => {
+            :validate => :number,
+            :default => 9 * 1024 * 1024 # 9MiB
+        },
+
         # Pass a set of key value pairs as the URL query string. This query string is added
         # to every host listed in the 'hosts' configuration. If the 'hosts' list contains
         # urls that already have query strings, the one specified here will be appended.

spec/integration/outputs/index_spec.rb

@@ -10,8 +10,7 @@
 require_relative "../../../spec/opensearch_spec_helper"
 require "logstash/outputs/opensearch"
 
-describe "TARGET_BULK_BYTES", :integration => true do
-  let(:target_bulk_bytes) { LogStash::Outputs::OpenSearch::TARGET_BULK_BYTES }
+describe "target_bulk_bytes", :integration => true do
   let(:event_count) { 1000 }
   let(:events) { event_count.times.map { event }.to_a }
   let(:config) {
@@ -32,11 +31,11 @@
   end
 
   describe "batches that are too large for one" do
-    let(:event) { LogStash::Event.new("message" => "a " * (((target_bulk_bytes/2) / event_count)+1)) }
+    let(:event) { LogStash::Event.new("message" => "a " * (((subject.client.target_bulk_bytes/2) / event_count)+1)) }
 
     it "should send in two batches" do
       expect(subject.client).to have_received(:bulk_send).twice do |payload|
-        expect(payload.size).to be <= target_bulk_bytes
+        expect(payload.size).to be <= subject.client.target_bulk_bytes
       end
     end
 
@@ -47,7 +46,7 @@
 
       it "should send in one batch" do
         expect(subject.client).to have_received(:bulk_send).once do |payload|
-          expect(payload.size).to be <= target_bulk_bytes
+          expect(payload.size).to be <= subject.client.target_bulk_bytes
         end
       end
     end

spec/unit/outputs/opensearch/http_client_spec.rb

@@ -17,6 +17,7 @@
     opts = {
       :hosts => [::LogStash::Util::SafeURI.new("127.0.0.1")],
       :logger => Cabin::Channel.get,
+      :target_bulk_bytes => 9_000_000,
       :metric => ::LogStash::Instrument::NullMetric.new(:dummy).namespace(:alsodummy)
     }
 
@@ -226,8 +227,8 @@
           end
         end
 
-        context "if a message is over TARGET_BULK_BYTES" do
-          let(:target_bulk_bytes) { LogStash::Outputs::OpenSearch::TARGET_BULK_BYTES }
+        context "if a message is over target_bulk_bytes" do
+          let(:target_bulk_bytes) { subject.target_bulk_bytes }
           let(:message) { "a" * (target_bulk_bytes + 1) }
 
           it "should be handled properly" do
@@ -256,8 +257,8 @@
             s = subject.send(:bulk, actions)
           end
 
-          context "if one exceeds TARGET_BULK_BYTES" do
-            let(:target_bulk_bytes) { LogStash::Outputs::OpenSearch::TARGET_BULK_BYTES }
+          context "if one exceeds target_bulk_bytes" do
+            let(:target_bulk_bytes) { subject.target_bulk_bytes }
             let(:message1) { "a" * (target_bulk_bytes + 1) }
             it "executes two bulk_send operations" do
               allow(subject).to receive(:join_bulk_responses)

spec/unit/outputs/opensearch_spec.rb

@@ -325,7 +325,7 @@
   end
 
   context '413 errors' do
-    let(:payload_size) { LogStash::Outputs::OpenSearch::TARGET_BULK_BYTES + 1024 }
+    let(:payload_size) { subject.client.target_bulk_bytes + 1024 }
     let(:event) { ::LogStash::Event.new("message" => ("a" * payload_size ) ) }
 
     let(:logger_stub) { double("logger").as_null_object }
@@ -357,7 +357,7 @@
 
       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))
+                                       hash_including(:action_count => 1, :content_length => a_value > subject.client.target_bulk_bytes))
     end
   end