Backport bulk count and uncompressed size headers to 11.x (+ ci/test fixes) (#1219)

Adds two new headers to each bulk request:

* "X-Elastic-Event-Count": number of actions / documents in that bulk request
* "X-Elastic-Uncompressed-Request-Length": size in bytes of the request body before compression

X-Elastic-Uncompressed-Request-Length is equal to Content-Length when compression is disabled.

Backport fixes:

* Tolerate the elasticsearch-ruby v8 client in integration tests. (#1208)
* elasticsearch-ruby client got updated to v8 in LS core. This plugin uses it in integration tests. This change tolerates both elasticsearch-ruby v7 and v8 client versions.
* Fix the ILM spec issue where  method was removed, restored internally.

Co-authored-by: Mashhur <[email protected]>

Author: jsvd

.ci/docker-compose.override.yml

@@ -1,5 +1,3 @@
-version: '3'
-
 services:
 
   logstash:

.ci/logstash-run.sh

@@ -13,7 +13,7 @@ else
 fi
 
 # CentOS 7 using curl defaults does not enable TLSv1.3
-CURL_OPTS="-k --tlsv1.2 --tls-max 1.3"
+CURL_OPTS="-k -u admin:elastic --tlsv1.2 --tls-max 1.3"
 
 wait_for_es() {
   count=120
@@ -22,7 +22,7 @@ wait_for_es() {
     [[ $count -eq 0 ]] && exit 1
     sleep 1
   done
-  echo $(curl $CURL_OPTS -vi $ES_URL | jq -r .version.number)
+  echo $(curl $CURL_OPTS $ES_URL | jq -r .version.number)
 }
 
 if [[ "$INTEGRATION" != "true" ]]; then

.travis.yml

@@ -4,13 +4,23 @@ import:
 jobs:
   include:
     - stage: "Integration Tests"
-      env: INTEGRATION=true SNAPSHOT=true LOG_LEVEL=info ELASTIC_STACK_VERSION=7.current
+      env: INTEGRATION=true LOG_LEVEL=info ELASTIC_STACK_VERSION=7.current
+    - env: INTEGRATION=true LOG_LEVEL=info ELASTIC_STACK_VERSION=8.previous
+    - env: INTEGRATION=true LOG_LEVEL=info ELASTIC_STACK_VERSION=8.current
+    - env: INTEGRATION=true LOG_LEVEL=info ELASTIC_STACK_VERSION=9.current
+    - env: INTEGRATION=true SNAPSHOT=true LOG_LEVEL=info ELASTIC_STACK_VERSION=7.current
     - env: INTEGRATION=true SNAPSHOT=true LOG_LEVEL=info ELASTIC_STACK_VERSION=8.previous
     - env: INTEGRATION=true SNAPSHOT=true LOG_LEVEL=info ELASTIC_STACK_VERSION=8.current
     - env: INTEGRATION=true SNAPSHOT=true LOG_LEVEL=info ELASTIC_STACK_VERSION=8.next
-    - env: INTEGRATION=true SNAPSHOT=true LOG_LEVEL=info ELASTIC_STACK_VERSION=8.future
+    - env: INTEGRATION=true SNAPSHOT=true LOG_LEVEL=info ELASTIC_STACK_VERSION=9.next
+    - env: INTEGRATION=true SNAPSHOT=true LOG_LEVEL=info ELASTIC_STACK_VERSION=main
     - stage: "Secure Integration Tests"
-      env: SECURE_INTEGRATION=true INTEGRATION=true LOG_LEVEL=info ELASTIC_STACK_VERSION=8.current SNAPSHOT=true
-    - env: SECURE_INTEGRATION=true INTEGRATION=true LOG_LEVEL=info ELASTIC_STACK_VERSION=7.current
-    - env: SECURE_INTEGRATION=true INTEGRATION=true LOG_LEVEL=info ELASTIC_STACK_VERSION=7.current ES_SSL_KEY_INVALID=true
-    - env: SECURE_INTEGRATION=true INTEGRATION=true LOG_LEVEL=info ELASTIC_STACK_VERSION=7.current ES_SSL_SUPPORTED_PROTOCOLS=TLSv1.3
+      env: SECURE_INTEGRATION=true INTEGRATION=true LOG_LEVEL=info ELASTIC_STACK_VERSION=7.current
+    - env: SECURE_INTEGRATION=true INTEGRATION=true LOG_LEVEL=info ELASTIC_STACK_VERSION=8.current
+    - env: SECURE_INTEGRATION=true INTEGRATION=true LOG_LEVEL=info ELASTIC_STACK_VERSION=8.current ES_SSL_KEY_INVALID=true
+    - env: SECURE_INTEGRATION=true INTEGRATION=true LOG_LEVEL=info ELASTIC_STACK_VERSION=8.current ES_SSL_SUPPORTED_PROTOCOLS=TLSv1.3
+    - env: SECURE_INTEGRATION=true INTEGRATION=true LOG_LEVEL=info ELASTIC_STACK_VERSION=9.current
+    - env: SECURE_INTEGRATION=true INTEGRATION=true LOG_LEVEL=info ELASTIC_STACK_VERSION=9.current ES_SSL_KEY_INVALID=true
+    - env: SECURE_INTEGRATION=true INTEGRATION=true LOG_LEVEL=info ELASTIC_STACK_VERSION=9.current ES_SSL_SUPPORTED_PROTOCOLS=TLSv1.3
+    - env: SECURE_INTEGRATION=true INTEGRATION=true LOG_LEVEL=info SNAPSHOT=true ELASTIC_STACK_VERSION=8.next
+    - env: SECURE_INTEGRATION=true INTEGRATION=true LOG_LEVEL=info SNAPSHOT=true ELASTIC_STACK_VERSION=9.next

CHANGELOG.md

@@ -1,5 +1,9 @@
+## 11.22.13
+ - Add headers reporting uncompressed size and doc count for bulk requests [#1217](https://github.com/logstash-plugins/logstash-output-elasticsearch/pull/1217)
+
 ## 11.22.12
  - Properly handle http code 413 (Payload Too Large) [#1199](https://github.com/logstash-plugins/logstash-output-elasticsearch/pull/1199)
+
 ## 11.22.11
  - Remove irrelevant log warning about elastic stack version [#1202](https://github.com/logstash-plugins/logstash-output-elasticsearch/pull/1202)
 

lib/logstash/outputs/elasticsearch/http_client.rb

@@ -21,7 +21,8 @@ module LogStash; module Outputs; class ElasticSearch;
   # 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
-
+  EVENT_COUNT_HEADER = "X-Elastic-Event-Count".freeze
+  UNCOMPRESSED_LENGTH_HEADER = "X-Elastic-Uncompressed-Request-Length".freeze
 
   class HttpClient
     attr_reader :client, :options, :logger, :pool, :action_count, :recv_count
@@ -143,7 +144,11 @@ def bulk(actions)
                        :payload_size => stream_writer.pos,
                        :content_length => body_stream.size,
                        :batch_offset => (index + 1 - batch_actions.size))
-          bulk_responses << bulk_send(body_stream, batch_actions)
+          headers = {
+            EVENT_COUNT_HEADER => batch_actions.size.to_s,
+            UNCOMPRESSED_LENGTH_HEADER => stream_writer.pos.to_s
+          }
+          bulk_responses << bulk_send(body_stream, batch_actions, headers)
           body_stream.truncate(0) && body_stream.seek(0)
           stream_writer = gzip_writer(body_stream) if compression_level?
           batch_actions.clear
@@ -159,7 +164,14 @@ def bulk(actions)
                    :payload_size => stream_writer.pos,
                    :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
+
+      if body_stream.size > 0
+        headers = {
+          EVENT_COUNT_HEADER => batch_actions.size.to_s,
+          UNCOMPRESSED_LENGTH_HEADER => stream_writer.pos.to_s
+        }
+        bulk_responses << bulk_send(body_stream, batch_actions, headers)
+      end
 
       body_stream.close unless compression_level?
       join_bulk_responses(bulk_responses)
@@ -179,8 +191,8 @@ def join_bulk_responses(bulk_responses)
       }
     end
 
-    def bulk_send(body_stream, batch_actions)
-      params = compression_level? ? {:headers => {"Content-Encoding" => "gzip"}} : {}
+    def bulk_send(body_stream, batch_actions, headers = {})
+      params = compression_level? ? {:headers => headers.merge("Content-Encoding" => "gzip") } : { :headers => headers }
 
       begin
         response = @pool.post(@bulk_path, params, body_stream.string)

logstash-output-elasticsearch.gemspec

@@ -1,6 +1,6 @@
 Gem::Specification.new do |s|
   s.name            = 'logstash-output-elasticsearch'
-  s.version         = '11.22.12'
+  s.version         = '11.22.13'
   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/es_spec_helper.rb

@@ -1,15 +1,18 @@
 require_relative './spec_helper'
 
 require 'elasticsearch'
-require_relative "support/elasticsearch/api/actions/delete_ilm_policy"
-require_relative "support/elasticsearch/api/actions/get_alias"
-require_relative "support/elasticsearch/api/actions/put_alias"
-require_relative "support/elasticsearch/api/actions/get_ilm_policy"
-require_relative "support/elasticsearch/api/actions/put_ilm_policy"
 
 require 'json'
 require 'cabin'
 
+# remove this condition and support package once plugin starts consuming elasticsearch-ruby v8 client
+# in elasticsearch-ruby v7, ILM APIs were in a separate xpack gem, now directly available
+unless elastic_ruby_v8_client_available?
+  require_relative "support/elasticsearch/api/actions/delete_ilm_policy"
+  require_relative "support/elasticsearch/api/actions/get_ilm_policy"
+  require_relative "support/elasticsearch/api/actions/put_ilm_policy"
+end
+
 module ESHelper
   def get_host_port
     if ENV["INTEGRATION"] == "true"
@@ -20,8 +23,12 @@ def get_host_port
   end
 
   def get_client
-    Elasticsearch::Client.new(:hosts => [get_host_port]).tap do |client|
-      allow(client).to receive(:verify_elasticsearch).and_return(true) # bypass client side version checking
+    if elastic_ruby_v8_client_available?
+      Elasticsearch::Client.new(:hosts => [get_host_port])
+    else
+      Elasticsearch::Client.new(:hosts => [get_host_port]).tap do |client|
+        allow(client).to receive(:verify_elasticsearch).and_return(true) # bypass client side version checking
+      end
     end
   end
 
@@ -128,31 +135,36 @@ def get_cluster_settings(client)
   end
 
   def get_policy(client, policy_name)
-    client.get_ilm_policy(name: policy_name)
+    if elastic_ruby_v8_client_available?
+      client.index_lifecycle_management.get_lifecycle(policy: policy_name)
+    else
+      client.get_ilm_policy(name: policy_name)
+    end
   end
 
   def put_policy(client, policy_name, policy)
-    client.put_ilm_policy({:name => policy_name, :body=> policy})
-  end
-
-  def put_alias(client, the_alias, index)
-    body = {
-        "aliases" => {
-            index => {
-                "is_write_index"=>  true
-            }
-        }
-    }
-    client.put_alias({name: the_alias, body: body})
+    if elastic_ruby_v8_client_available?
+      client.index_lifecycle_management.put_lifecycle({:policy => policy_name, :body=> policy})
+    else
+      client.put_ilm_policy({:name => policy_name, :body=> policy})
+    end
   end
 
   def clean_ilm(client)
-    client.get_ilm_policy.each_key { |key| client.delete_ilm_policy(name: key)  if key =~ /logstash-policy/ }
+    if elastic_ruby_v8_client_available?
+      client.index_lifecycle_management.get_lifecycle.each_key { |key| client.index_lifecycle_management.delete_lifecycle(policy: key) if key =~ /logstash-policy/ }
+    else
+      client.get_ilm_policy.each_key { |key| client.delete_ilm_policy(name: key) if key =~ /logstash-policy/ }
+    end
   end
 
   def supports_ilm?(client)
     begin
-      client.get_ilm_policy
+      if elastic_ruby_v8_client_available?
+        client.index_lifecycle_management.get_lifecycle
+      else
+        client.get_ilm_policy
+      end
       true
     rescue
       false

spec/fixtures/test_certs/GENERATED_AT

@@ -1 +1 @@
-2024-06-25T21:50:58+01:00
+2025-07-22T11:15:03+01:00

spec/fixtures/test_certs/ca.crt

@@ -1,29 +1,32 @@
 -----BEGIN CERTIFICATE-----
-MIIFDDCCAvQCAQEwDQYJKoZIhvcNAQELBQAwTDELMAkGA1UEBhMCUFQxCzAJBgNV
-BAgMAk5BMQ8wDQYDVQQHDAZMaXNib24xDjAMBgNVBAoMBU15TGFiMQ8wDQYDVQQD
-DAZSb290Q0EwHhcNMjQwNjI1MjA1MDU4WhcNMjUwNjI1MjA1MDU4WjBMMQswCQYD
-VQQGEwJQVDELMAkGA1UECAwCTkExDzANBgNVBAcMBkxpc2JvbjEOMAwGA1UECgwF
-TXlMYWIxDzANBgNVBAMMBlJvb3RDQTCCAiIwDQYJKoZIhvcNAQEBBQADggIPADCC
-AgoCggIBAMtTMqAWuH17b9XqPa5L3HNqgnZ958+gvcOt7Q/sOEvcDQJgkzZ+Gywh
-5er5JF2iomYOHiD5JncYr4YmRQKuYfD6B1WI5FuQthD/OlA1/RHqtbY27J33SaO6
-6ro8gal7vjHrXKQkefVYRwdfO6DqqbhV6L4sMiy8FzQ55TMpoM35cWuvoAMxvSQq
-GZ4pYYKnfNSGhzHvssfNS1xu/Lwb7Vju4jPhp+43BkGwEimI5km7jNC1nwjiHtxD
-sY/s93AKa/vLktXKUK5nA3jjJOhAbRTVnbOAgxFt0YbX98xW/aUqscgBUVs9J/My
-TRMwVKJ7Vsmth1PdJQksUASuzESlSPl09dMjTQ+MXzJDt0JvX8SIJPmbBng78MSa
-CUhpOZiii1l2mBfPWejx20I/SMCUNmzbwm2w9JD50Jv2iX4l4ge4H1CIK1/orW1p
-dY9xPL0uKYm6ADsDC0B8sGgNMBXeB6aLojY1/ITwmmfpfk9c/yWPfC7stHgCYRAv
-5MfGAsmv0/ya5VrWQGBJkFiYy1pon6nxUjCbgn0RABojRoGdhhY3QDipgwmSgFZx
-r064RFr1bt/Ml3MJmPf535mSwPdk/j/zw4IZTvlmwKW3FyMDhwYL/zX7J0c6MzMP
-LEdi73Qjzmr3ENIrir4O86wNz81YRfYkg9ZX8yKJK9LBAUrYCjJ3AgMBAAEwDQYJ
-KoZIhvcNAQELBQADggIBABym9LMyS9W9lvpcH4OK1YLfBPJwrhZ+4keiriY4zWOo
-pB+v2Q35neMMXSlTDpeIwPdMkqsh8VZprOWURF80JGvpJ6fBfi05rCDWp/ol1ZKi
-snCA+dE2zDK7Z3+F0MbakT5oBi5WgkXSvRvlJEJ/gBD7WC1wq0kxCMK+M5w2RPAT
-nnV/iozNBkwExxyJA7BpS6F/v0XjwK7fm5Kpql7zKlh5piZ2IVU0B60Sqskcb2mU
-90+1r9T06ekIW/Iz1jd5RWYziu0nbmDeKeKvGAICNU+evYXW+/5kKecMLuEvDCgS
-ssbt/Hb510uLHhxfhN4SbvBl2zADsLC+2arf2ATIwD8ZXDDs04ayBsejV0ZwVrTZ
-ExKqAys+B3tuIHGRqL5VukdmH6g6oJziYueohPBCOuSOzDd0FhppF4uXZS8DReSg
-KieO2ZYfiA1gVRiY6jPx+r7J9I5kSS1gwr/e3zHJHa79ijMB1SSIswQUmgSMkwGh
-sNyDNI9ZxgJan3v7kVargMt2LiNcXvVyTzPSYSXcY7SoebfpMprVIG7vZ9TZf+Uu
-FQeOfxdLFuGTnpFrYmvOD3OIKfODlY5t+TNICg7A3eTUXeJPcdBBnuVCiQU6TCB5
-H+69K5w54Q6a70sHZU1IWsGT8XtbUizPNQky+LAFsE/5oUnCqtypeEu4srcZK53x
+MIIFdTCCA12gAwIBAgIUDITbsLT9hKser0ZzBZsxqgaZdWswDQYJKoZIhvcNAQEL
+BQAwSjELMAkGA1UEBhMCUFQxCzAJBgNVBAgMAk5BMQ8wDQYDVQQHDAZMaXNib24x
+DjAMBgNVBAoMBU15TGFiMQ0wCwYDVQQDDARyb290MB4XDTI1MDcyMjEwMTUwM1oX
+DTM1MDcyMDEwMTUwM1owSjELMAkGA1UEBhMCUFQxCzAJBgNVBAgMAk5BMQ8wDQYD
+VQQHDAZMaXNib24xDjAMBgNVBAoMBU15TGFiMQ0wCwYDVQQDDARyb290MIICIjAN
+BgkqhkiG9w0BAQEFAAOCAg8AMIICCgKCAgEAy1MyoBa4fXtv1eo9rkvcc2qCdn3n
+z6C9w63tD+w4S9wNAmCTNn4bLCHl6vkkXaKiZg4eIPkmdxivhiZFAq5h8PoHVYjk
+W5C2EP86UDX9Eeq1tjbsnfdJo7rqujyBqXu+MetcpCR59VhHB187oOqpuFXoviwy
+LLwXNDnlMymgzflxa6+gAzG9JCoZnilhgqd81IaHMe+yx81LXG78vBvtWO7iM+Gn
+7jcGQbASKYjmSbuM0LWfCOIe3EOxj+z3cApr+8uS1cpQrmcDeOMk6EBtFNWds4CD
+EW3Rhtf3zFb9pSqxyAFRWz0n8zJNEzBUontWya2HU90lCSxQBK7MRKVI+XT10yNN
+D4xfMkO3Qm9fxIgk+ZsGeDvwxJoJSGk5mKKLWXaYF89Z6PHbQj9IwJQ2bNvCbbD0
+kPnQm/aJfiXiB7gfUIgrX+itbWl1j3E8vS4piboAOwMLQHywaA0wFd4HpouiNjX8
+hPCaZ+l+T1z/JY98Luy0eAJhEC/kx8YCya/T/JrlWtZAYEmQWJjLWmifqfFSMJuC
+fREAGiNGgZ2GFjdAOKmDCZKAVnGvTrhEWvVu38yXcwmY9/nfmZLA92T+P/PDghlO
++WbApbcXIwOHBgv/NfsnRzozMw8sR2LvdCPOavcQ0iuKvg7zrA3PzVhF9iSD1lfz
+Iokr0sEBStgKMncCAwEAAaNTMFEwHQYDVR0OBBYEFKFadJx46upif1BrhYZ0iu8o
+2z8rMB8GA1UdIwQYMBaAFKFadJx46upif1BrhYZ0iu8o2z8rMA8GA1UdEwEB/wQF
+MAMBAf8wDQYJKoZIhvcNAQELBQADggIBAJi4FwYJz/RotoUpfrLZFf69RoI01Fje
+8ITt8SR1Dx/1GTPEuqVVfx0EYtOoH6Gg3FwgSQ9GHRDIa1vkHY5S+FUSOW3pCoZE
+/kaLu9bmFxn+GntghvQEor+LzODuZKLXupaGcu1tA4fzyuI4jglVD2sGZtLk//CT
+Hd4tOWXo5k1Fj0jMnJq+2Htr8yBeSAO5ZNsvtAjOUU6pfDEwL9bgRzlKKFQQMUYo
+6x1FvRDRXWjpzB/H+OSqOaoNLEB9FfEl8I7nn6uTenr5WxjPAOpwjZl9ObB/95xM
+p91abKbLQLev5I8npM9G3C/n01l3IzRs7DNHqGJTZO7frGhicD7/jNa+tkSioeJ2
+fIMqgDOvQE+gMxs19zw1tsI3+kqX7+ptTkU4Lan5V5ZKGfU8xtcVIlyRk5/yDUI5
+1dfQVubs6z07s6De2qa92LFz9l8sT6QuVer+c/wPPhBdMwbzcHyUJIBjFaBpxH86
+F7Mr5Zr/+qcbHglAHow1lBqdZzimqGd1koqFRat/pFUFh0iqktMmpl+ZUCjyoQEX
+93j8aMU2UQjYM8NJDE2aRculo9OEoqERYFM2m3nHvrtE7iZgddryLNH7ZmC1EquX
+MhZJ26GuZ2U4b9dAX858WTv0q1EF5S8KObMlxMU7IDk+cWlSD+puWliwfUKoTR/4
+JErSfjCSaRqh
 -----END CERTIFICATE-----

spec/fixtures/test_certs/ca.der.sha256

@@ -1 +1 @@
-8b23238088af65cbae6ee9c23821068d896ec1dad081e2a1035ff70866943247
+d403930d5296f1515aadd3f730757e7719188b63a276687a3475128b746e4340