Add http compression level (#1148)

This commit deprecated `http_compression` in favor of `compression_level`. http_compression `false` mapped to compression_level `0`, and `true` mapped to `1`. The default setting has changed from disabling compression to enabling compression at level 1 (best speed) Co-authored-by: Edmo Vamerlatti Costa <11836452+edmocosta@users.noreply.github.com>

Author: kaisecheng

CHANGELOG.md

@@ -1,3 +1,6 @@
+## 11.17.0
+ - Added support to http compression level. Deprecated `http_compression` in favour of `compression_level` and enabled compression level 1 by default. [#1148](https://github.com/logstash-plugins/logstash-output-elasticsearch/pull/1148)
+
 ## 11.16.0
  - Added support to Serverless Elasticsearch [#1445](https://github.com/logstash-plugins/logstash-output-elasticsearch/pull/1145)
 

docs/index.asciidoc

@@ -277,9 +277,9 @@ not reevaluate its DNS value while the keepalive is in effect.
 ==== HTTP Compression
 
 This plugin always reads compressed responses from {es}.
-It _can be configured_ to send compressed bulk requests to {es}.
+By default, it sends compressed bulk requests to {es}.
 
-If you are concerned about bandwidth, you can enable <<plugins-{type}s-{plugin}-http_compression>> to trade a small amount of CPU capacity for a significant reduction in network IO.
+If you are concerned about bandwidth, you can set a higher <<plugins-{type}s-{plugin}-compression_level>> to trade CPU capacity for a reduction in network IO.
 
 ==== Authentication
 
@@ -310,6 +310,7 @@ This plugin supports the following configuration options plus the
 | <<plugins-{type}s-{plugin}-ca_trusted_fingerprint>> |<<string,string>>|No
 | <<plugins-{type}s-{plugin}-cloud_auth>> |<<password,password>>|No
 | <<plugins-{type}s-{plugin}-cloud_id>> |<<string,string>>|No
+| <<plugins-{type}s-{plugin}-compression_level>> |<<number,number>>, one of `[0 ~ 9]`|No
 | <<plugins-{type}s-{plugin}-custom_headers>> |<<hash,hash>>|No
 | <<plugins-{type}s-{plugin}-data_stream>> |<<string,string>>, one of `["true", "false", "auto"]`|No
 | <<plugins-{type}s-{plugin}-data_stream_auto_routing>> |<<boolean,boolean>>|No
@@ -459,6 +460,17 @@ Cloud ID, from the Elastic Cloud web console. If set `hosts` should not be used.
 For more details, check out the
 {logstash-ref}/connecting-to-cloud.html[Logstash-to-Cloud documentation].
 
+[id="plugins-{type}s-{plugin}-compression_level"]
+===== `compression_level`
+
+ * Value can be any of: `0`, `1`, `2`, `3`, `4`, `5`, `6`, `7`, `8`, `9`
+ * Default value is `1`
+
+The gzip compression level. Setting this value to `0` disables compression.
+The compression level must be in the range of `1` (best speed) to `9` (best compression).
+
+Increasing the compression level will reduce the network usage but will increase the CPU usage.
+
 [id="plugins-{type}s-{plugin}-data_stream"]
 ===== `data_stream`
 
@@ -618,7 +630,7 @@ NOTE: Deprecated, refer to <<plugins-{type}s-{plugin}-silence_errors_in_log>>.
 Pass a set of key value pairs as the headers sent in each request to
 an elasticsearch node. The headers will be used for any kind of request
 (_bulk request, template installation, health checks and sniffing).
-These custom headers will be overidden by settings like `http_compression`.
+These custom headers will be overidden by settings like `compression_level`.
 
 [id="plugins-{type}s-{plugin}-healthcheck_path"]
 ===== `healthcheck_path`
@@ -659,11 +671,12 @@ Any special characters present in the URLs here MUST be URL escaped! This means
 
 [id="plugins-{type}s-{plugin}-http_compression"]
 ===== `http_compression`
+deprecated[11.17.0, Replaced by <<plugins-{type}s-{plugin}-compression_level>>]
 
  * Value type is <<boolean,boolean>>
  * Default value is `false`
 
-Enable gzip compression on requests.
+Setting `true` enables gzip compression level 1 on requests.
 
 This setting allows you to reduce this plugin's outbound network traffic by
 compressing each bulk _request_ to {es}.

lib/logstash/outputs/elasticsearch.rb

@@ -276,6 +276,7 @@ def initialize(*params)
  super
  setup_ecs_compatibility_related_defaults
  setup_ssl_params!
+ setup_compression_level!
  end
 
  def register
@@ -368,6 +369,7 @@ def config_init(params)
  params['proxy'] = proxy # do not do resolving again
  end
  end
+
  super(params)
  end
 
@@ -669,6 +671,20 @@ def setup_ssl_params!
  params['ssl_verification_mode'] = @ssl_verification_mode unless @ssl_verification_mode.nil?
  end
 
+ def setup_compression_level!
+ @compression_level = normalize_config(:compression_level) do |normalize|
+ normalize.with_deprecated_mapping(:http_compression) do |http_compression|
+ if http_compression == true
+ DEFAULT_ZIP_LEVEL
+ else
+ 0
+ end
+ end
+ end
+
+ params['compression_level'] = @compression_level unless @compression_level.nil?
+ end
+
  # To be overidden by the -java version
  VALID_HTTP_ACTIONS = ["index", "delete", "create", "update"]
  def valid_actions

lib/logstash/outputs/elasticsearch/http_client.rb

@@ -118,7 +118,7 @@ def bulk(actions)
  end
 
  body_stream = StringIO.new
- if http_compression
+ if compression_level?
  body_stream.set_encoding "BINARY"
  stream_writer = gzip_writer(body_stream)
  else
@@ -141,14 +141,14 @@ def bulk(actions)
  :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
+ stream_writer = gzip_writer(body_stream) if compression_level?
  batch_actions.clear
  end
  stream_writer.write(as_json)
  batch_actions << action
  end
 
- stream_writer.close if http_compression
+ stream_writer.close if compression_level?
 
  logger.debug("Sending final bulk request for batch.",
  :action_count => batch_actions.size,
@@ -157,15 +157,15 @@ def bulk(actions)
  :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
+ body_stream.close unless compression_level?
  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)
+ Zlib::GzipWriter.new(io, client_settings.fetch(:compression_level), Zlib::DEFAULT_STRATEGY)
  end
 
  def join_bulk_responses(bulk_responses)
@@ -176,7 +176,7 @@ def join_bulk_responses(bulk_responses)
  end
 
  def bulk_send(body_stream, batch_actions)
- params = http_compression ? {:headers => {"Content-Encoding" => "gzip"}} : {}
+ params = compression_level? ? {:headers => {"Content-Encoding" => "gzip"}} : {}
  response = @pool.post(@bulk_path, params, body_stream.string)
 
  @bulk_response_metrics.increment(response.code.to_s)
@@ -298,8 +298,10 @@ def ssl_options
  @_ssl_options ||= client_settings.fetch(:ssl, {})
  end
 
- def http_compression
- client_settings.fetch(:http_compression, false)
+ # return true if compression_level is [1..9]
+ # return false if it is 0
+ def compression_level?
+ client_settings.fetch(:compression_level) > 0
  end
 
  def build_adapter(options)

lib/logstash/outputs/elasticsearch/http_client_builder.rb

@@ -8,7 +8,7 @@ def self.build(logger, hosts, params)
  :pool_max => params["pool_max"],
  :pool_max_per_route => params["pool_max_per_route"],
  :check_connection_timeout => params["validate_after_inactivity"],
- :http_compression => params["http_compression"],
+ :compression_level => params["compression_level"],
  :headers => params["custom_headers"] || {}
  }
 

lib/logstash/plugin_mixins/elasticsearch/api_configs.rb

@@ -7,6 +7,7 @@ module APIConfigs
  # This module defines common options that can be reused by alternate elasticsearch output plugins such as the elasticsearch_data_streams output.
 
  DEFAULT_HOST = ::LogStash::Util::SafeURI.new("//127.0.0.1")
+ DEFAULT_ZIP_LEVEL = 1
 
  CONFIG_PARAMS = {
  # Username to authenticate to a secure Elasticsearch cluster
@@ -186,7 +187,14 @@ module APIConfigs
  :validate_after_inactivity => { :validate => :number, :default => 10000 },
 
  # Enable gzip compression on requests. Note that response compression is on by default for Elasticsearch v5.0 and beyond
- :http_compression => { :validate => :boolean, :default => false },
+ # Set `true` to enable compression with level 1
+ # Set `false` to disable compression with level 0
+ :http_compression => { :validate => :boolean, :default => true, :deprecated => "Set 'compression_level' instead." },
+
+ # Number `1` ~ `9` are the gzip compression level
+ # Set `0` to disable compression
+ # Set `1` (best speed) to `9` (best compression) to use compression
+ :compression_level => { :validate => [ 0, 1, 2, 3, 4, 5, 6, 7, 8, 9 ], :default => DEFAULT_ZIP_LEVEL },
 
  # Custom Headers to send on each request to elasticsearch nodes
  :custom_headers => { :validate => :hash, :default => {} },

logstash-output-elasticsearch.gemspec

@@ -1,6 +1,6 @@
 Gem::Specification.new do |s|
  s.name = 'logstash-output-elasticsearch'
- s.version = '11.16.0'
+ s.version = '11.17.0'
  s.licenses = ['apache-2.0']
  s.summary = "Stores logs in Elasticsearch"
  s.description = "This gem is a Logstash plugin required to be installed on top of the Logstash core pipeline using $LS_HOME/bin/logstash-plugin install gemname. This gem is not a stand-alone program"

spec/integration/outputs/compressed_indexing_spec.rb

@@ -8,63 +8,64 @@
  }
 end
 
-describe "indexing with http_compression turned on", :integration => true do
- let(:event) { LogStash::Event.new("message" => "Hello World!", "type" => type) }
- let(:index) { 10.times.collect { rand(10).to_s }.join("") }
- let(:type) { ESHelper.es_version_satisfies?("< 7") ? "doc" : "_doc" }
- let(:event_count) { 10000 + rand(500) }
- let(:events) { event_count.times.map { event }.to_a }
- let(:config) {
- {
- "hosts" => get_host_port,
- "index" => index,
- "http_compression" => true
+[ {"http_compression" => true}, {"compression_level" => 1} ].each do |compression_config|
+ describe "indexing with http_compression turned on", :integration => true do
+ let(:event) { LogStash::Event.new("message" => "Hello World!", "type" => type) }
+ let(:index) { 10.times.collect { rand(10).to_s }.join("") }
+ let(:type) { ESHelper.es_version_satisfies?("< 7") ? "doc" : "_doc" }
+ let(:event_count) { 10000 + rand(500) }
+ let(:events) { event_count.times.map { event }.to_a }
+ let(:config) {
+ {
+ "hosts" => get_host_port,
+ "index" => index
+ }
  }
- }
- subject { LogStash::Outputs::ElasticSearch.new(config) }
+ subject { LogStash::Outputs::ElasticSearch.new(config.merge(compression_config)) }
 
- let(:es_url) { "http://#{get_host_port}" }
- let(:index_url) {"#{es_url}/#{index}"}
- let(:http_client_options) { {} }
- let(:http_client) do
- Manticore::Client.new(http_client_options)
- end
+  let(:es_url) { "http://#{get_host_port}" }
+  let(:index_url) {"#{es_url}/#{index}"}
+  let(:http_client_options) { {} }
+  let(:http_client) do
+  Manticore::Client.new(http_client_options)
+  end
 
- before do
- subject.register
- subject.multi_receive([])
- end
+  before do
+  subject.register
+  subject.multi_receive([])
+  end
 
- shared_examples "an indexer" do
- it "ships events" do
- subject.multi_receive(events)
+  shared_examples "an indexer" do
+  it "ships events" do
+  subject.multi_receive(events)
 
- http_client.post("#{es_url}/_refresh").call
+  http_client.post("#{es_url}/_refresh").call
 
- response = http_client.get("#{index_url}/_count?q=*")
- result = LogStash::Json.load(response.body)
- cur_count = result["count"]
- expect(cur_count).to eq(event_count)
+  response = http_client.get("#{index_url}/_count?q=*")
+  result = LogStash::Json.load(response.body)
+  cur_count = result["count"]
+  expect(cur_count).to eq(event_count)
 
- response = http_client.get("#{index_url}/_search?q=*&size=1000")
- result = LogStash::Json.load(response.body)
- result["hits"]["hits"].each do |doc|
- if ESHelper.es_version_satisfies?("< 8")
- expect(doc["_type"]).to eq(type)
- else
- expect(doc).not_to include("_type")
+ response = http_client.get("#{index_url}/_search?q=*&size=1000")
+ result = LogStash::Json.load(response.body)
+ result["hits"]["hits"].each do |doc|
+ if ESHelper.es_version_satisfies?("< 8")
+ expect(doc["_type"]).to eq(type)
+ else
+ expect(doc).not_to include("_type")
+ end
+ expect(doc["_index"]).to eq(index)
  end
- expect(doc["_index"]).to eq(index)
  end
  end
- end
 
- it "sets the correct content-encoding header and body is compressed" do
- expect(subject.client.pool.adapter.client).to receive(:send).
- with(anything, anything, {:headers=>{"Content-Encoding"=>"gzip", "Content-Type"=>"application/json"}, :body => a_valid_gzip_encoded_string}).
- and_call_original
- subject.multi_receive(events)
- end
+  it "sets the correct content-encoding header and body is compressed" do
+  expect(subject.client.pool.adapter.client).to receive(:send).
+  with(anything, anything, {:headers=>{"Content-Encoding"=>"gzip", "Content-Type"=>"application/json"}, :body => a_valid_gzip_encoded_string}).
+  and_call_original
+  subject.multi_receive(events)
+  end
 
- it_behaves_like("an indexer")
-end
+ it_behaves_like("an indexer")
+ end
+end

spec/integration/outputs/index_spec.rb

@@ -262,7 +262,8 @@ def curl_and_get_json_response(url, method: :get, retrieve_err_payload: false);
  let(:config) {
  {
  "hosts" => get_host_port,
- "index" => index
+ "index" => index,
+ "http_compression" => false
  }
  }
  it_behaves_like("an indexer")
@@ -273,7 +274,8 @@ def curl_and_get_json_response(url, method: :get, retrieve_err_payload: false);
  let(:config) {
  {
  "hosts" => get_host_port,
- "index" => index
+ "index" => index,
+ "http_compression" => false
  }
  }
  it_behaves_like("an indexer")
@@ -291,7 +293,8 @@ def curl_and_get_json_response(url, method: :get, retrieve_err_payload: false);
  "password" => password,
  "ssl_enabled" => true,
  "ssl_certificate_authorities" => cacert,
- "index" => index
+ "index" => index,
+ "http_compression" => false
  }
  end 
 
@@ -351,7 +354,8 @@ def curl_and_get_json_response(url, method: :get, retrieve_err_payload: false);
  "hosts" => ["https://#{CGI.escape(user)}:#{CGI.escape(password)}@elasticsearch:9200"],
  "ssl_enabled" => true,
  "ssl_certificate_authorities" => "spec/fixtures/test_certs/test.crt",
- "index" => index
+ "index" => index,
+ "http_compression" => false
  }
  end