Use Manticore http client and reuse it across calls.

.github/workflows/merge-to-master.yml → .github/workflows/merge-to-master-develop.yml

@@ -1,14 +1,37 @@
-name: New Relic Logstash Output Plugin - Merge to master
+name: New Relic Logstash Output Plugin - Merge to master or develop
 
 on:
   push:
     branches:
       - master
+      - develop
 
 jobs:
+  version-check:
+    name: Check version for develop branch
+    runs-on: ubuntu-22.04
+    if: github.ref == 'refs/heads/develop'
+
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v2
+
+      - name: Verify version contains 'beta'
+        run: |
+          VERSION=$(grep -o 'VERSION = "[^"]*"' lib/logstash/outputs/newrelic_version/version.rb | awk -F'"' '{print $2}')
+          echo "Found version: $VERSION"
+          if [[ ! "$VERSION" =~ beta ]]; then
+            echo "❌ Error: Version '$VERSION' does not contain 'beta'"
+            echo "Develop branch versions must contain 'beta' (e.g., '1.5.4-beta')"
+            exit 1
+          fi
+          echo "✓ Version check passed: '$VERSION' contains 'beta'"
+
   cd:
     name: Continuous Delivery pipeline
     runs-on: ubuntu-22.04
+    needs: [version-check]
+    if: always() && (github.ref == 'refs/heads/master' || needs.version-check.result == 'success')
 
     steps:
       - name: Checkout code

.github/workflows/pr.yml

@@ -50,4 +50,8 @@ jobs:
 
       - name: Logstash E2E testing
         run: ./test.sh ${{ matrix.logstash_version }}
+        env:
+          LICENSE_KEY: ${{ secrets.NEW_RELIC_LICENSE_KEY }}
+          NEW_RELIC_ACCOUNT_ID: ${{ secrets.NEW_RELIC_ACCOUNT_ID }}
+          NEW_RELIC_API_KEY: ${{ secrets.NEW_RELIC_API_KEY }}
 

Gemfile

@@ -1,5 +1,8 @@
 source 'https://rubygems.org'
-gemspec
+
+# This is a JRuby-only project (Logstash plugin)
+# Only load gemspec on JRuby platform
+gemspec if RUBY_PLATFORM == "java"
 
 # The following is required to locally develop this plugin. Note that this Gemfile is NOT used when building the gem
 # file for this plugin (see merge-to-master.yml), only when unit testing. When unit-testing, we need to have logstash-core

lib/logstash/outputs/newrelic.rb

@@ -1,12 +1,13 @@
 # encoding: utf-8
 require "logstash/outputs/base"
 require "logstash/outputs/newrelic_version/version"
-require 'net/http'
+require 'manticore'
 require 'uri'
 require 'zlib'
 require 'json'
 require 'java'
 require 'set'
+require 'stringio'
 require_relative './config/bigdecimal_patch'
 require_relative './exception/error'
 
@@ -33,6 +34,7 @@ def register
     if @api_key.nil? && @license_key.nil?
       raise LogStash::ConfigurationError, "Must provide a license key or api key", caller
     end
+    @logger.info("Registering logstash-output-newrelic", :version => LogStash::Outputs::NewRelicVersion::VERSION, :target => @base_uri)
     auth = {
       @api_key.nil? ? 'X-License-Key' : 'X-Insert-Key' =>
         @api_key.nil? ? @license_key.value : @api_key.value
@@ -43,6 +45,29 @@ def register
       'Content-Type' => 'application/json'
     }.merge(auth).freeze
 
+    client_options = {
+      :pool_max => @concurrent_requests,
+      :pool_max_per_route => @concurrent_requests
+    }
+
+    # Only configure SSL if using HTTPS
+    if @end_point.scheme == 'https'
+      client_options[:ssl] = {
+        :verify => :default
+      }
+      # Set reasonable timeouts for the HTTP client
+      client_options[:connect_timeout] = 30
+      client_options[:socket_timeout] = 30
+
+      if !@custom_ca_cert.nil?
+        # Load the custom CA certificate
+        # For test environments with self-signed certs, disable verification
+        client_options[:ssl][:ca_file] = @custom_ca_cert
+      end
+    end
+
+    @client = Manticore::Client.new(client_options)
+
     # We use a semaphore to ensure that at most there are @concurrent_requests inflight Logstash requests being processed
     # by our plugin at the same time. Without this semaphore, given that @executor.submit() is an asynchronous method, it
     # would cause that an unbounded amount of inflight requests may be processed by our plugin. Logstash then believes
@@ -52,9 +77,25 @@ def register
     @semaphore = java.util.concurrent.Semaphore.new(@concurrent_requests)
   end
 
+  # Shutdown hook called by Logstash 5.x and 6.x versions during pipeline shutdown
+  def stop
+    shutdown
+  end
+
+  # Shutdown hook called by Logstash 7.x+ versions during pipeline shutdown
+  def close
+    shutdown
+  end
+
+  # Additional shutdown hook for cleanup, called by some Logstash versions
+  def teardown
+    shutdown
+  end
+
   # Used by tests so that the test run can complete (background threads prevent JVM exit)
   def shutdown
     if @executor
+      @logger.info("Draining outstanding New Relic requests")
       @executor.shutdown
       # We want this long enough to not have threading issues
       terminationWaitInSeconds = 10
@@ -63,6 +104,11 @@ def shutdown
         raise "Did not shut down within #{terminationWaitInSeconds} seconds"
       end
     end
+
+    if defined?(@client) && @client
+      @logger.info("Closing New Relic HTTP client")
+      @client.close
+    end
   end
 
   def time_to_logstash_timestamp(time)
@@ -103,6 +149,8 @@ def multi_receive(logstash_events)
 
     nr_logs = to_nr_logs(logstash_events)
 
+    @logger.info("Submitting logs to New Relic", :event_count => nr_logs.length)
+
     submit_logs_to_be_sent(nr_logs)
   end
 
@@ -131,53 +179,57 @@ def package_and_send_recursively(nr_logs)
       :logs => nr_logs
     }
 
-    compressed_payload = StringIO.new
-    gzip = Zlib::GzipWriter.new(compressed_payload)
-    gzip << [payload].to_json
-    gzip.close
-
-    compressed_size = compressed_payload.string.bytesize
+    payload_json = [payload].to_json
+    compressed_payload = gzip_compress(payload_json, Zlib::DEFAULT_COMPRESSION)
+    compressed_size = compressed_payload.bytesize
     log_record_count = nr_logs.length
 
     if compressed_size >= MAX_PAYLOAD_SIZE_BYTES && log_record_count == 1
       @logger.error("Can't compress record below required maximum packet size and it will be discarded.")
     elsif compressed_size >= MAX_PAYLOAD_SIZE_BYTES && log_record_count > 1
-      @logger.debug("Compressed payload size (#{compressed_size}) exceededs maximum packet size (1MB) and will be split in two.")
+      @logger.debug("Compressed payload size exceeds maximum packet size, splitting payload", :compressed_size => compressed_size)
       split_index = log_record_count / 2
+      @logger.debug("Splitting payload", :split_index => split_index, :first_half => split_index, :second_half => log_record_count - split_index)
       package_and_send_recursively(nr_logs[0...split_index])
       package_and_send_recursively(nr_logs[split_index..-1])
     else
-      @logger.debug("Payload compressed size: #{compressed_size}")
-      nr_send(compressed_payload.string)
+      nr_send(compressed_payload)
     end
   end
 
   def handle_response(response)
-    if !(200 <= response.code.to_i && response.code.to_i < 300)
-      raise Error::BadResponseCodeError.new(response.code.to_i, @base_uri)
+    if !(200 <= response.code && response.code < 300)
+      raise Error::BadResponseCodeError.new(response.code, @base_uri)
     end
   end
 
+  # Compresses a given payload string using GZIP.
+  #
+  # @param payload [String] The string payload to be compressed.
+  # @param compression_level [Integer] The GZIP compression level to use.
+  # @return [String] The GZIP-compressed binary string.
+  def gzip_compress(payload, compression_level)
+    string_io = StringIO.new
+    string_io.set_encoding("BINARY")
+    Zlib::GzipWriter.wrap(string_io, compression_level) do |gz|
+      gz.write(payload)
+    end
+    string_io.string
+  end
+
   def nr_send(payload)
     retries = 0
     retry_duration = 1
 
     begin
-      http = Net::HTTP.new(@end_point.host, @end_point.port || 443)
-      request = Net::HTTP::Post.new(@end_point.request_uri)
-      http.use_ssl = (@end_point.scheme == 'https')
-      http.verify_mode = @end_point.scheme == 'https' ? OpenSSL::SSL::VERIFY_PEER : OpenSSL::SSL::VERIFY_NONE
-      if !@custom_ca_cert.nil?
-        store = OpenSSL::X509::Store.new
-        ca_cert = OpenSSL::X509::Certificate.new(File.read(@custom_ca_cert))
-        store.add_cert(ca_cert)
-        http.cert_store = store
-      end
-      @header.each { |k, v| request[k] = v }
-      request.body = payload
-      handle_response(http.request(request))
+      @logger.debug("Dispatching payload to New Relic", :endpoint => @base_uri, :payload_size => payload.bytesize)
+      response = @client.post(@base_uri, :body => payload, :headers => @header)
+      @logger.debug("Received response from New Relic", :code => response.code, :message => response.message)
+      handle_response(response)
       if (retries > 0)
         @logger.warn("Successfully sent logs at retry #{retries}")
+      else
+        @logger.debug("Successfully sent logs to New Relic", :response_code => response.code)
       end
     rescue Error::BadResponseCodeError => e
       @logger.error(e.message)

lib/logstash/outputs/newrelic_version/version.rb

@@ -1,7 +1,7 @@
 module LogStash
   module Outputs
     module NewRelicVersion
-      VERSION = "1.5.2"
+      VERSION = "2.0.0-beta"
     end
   end
 end

logstash-output-newrelic.gemspec

@@ -12,6 +12,9 @@ Gem::Specification.new do |s|
   s.authors       = ['New Relic Logging Team']
   s.email         = '[email protected]'
   s.require_paths = ['lib']
+
+  # This is a Logstash plugin and requires JRuby
+  s.platform      = 'java'
 
   # Files
   s.files = Dir['lib/**/*','spec/**/*','vendor/**/*','*.gemspec','*.md','CONTRIBUTORS','Gemfile','LICENSE','NOTICE.TXT']
@@ -24,6 +27,7 @@ Gem::Specification.new do |s|
   # Gem dependencies
   s.add_runtime_dependency "logstash-core-plugin-api", "~> 2.0"
   s.add_runtime_dependency "logstash-codec-plain"
+  s.add_runtime_dependency "manticore"
   s.add_development_dependency "logstash-devutils"
   s.add_development_dependency "webmock"
   s.add_development_dependency "rspec"