ssl reloading: pull ownership of rebuilding into es client holder

x-pack/lib/config_management/elasticsearch_source.rb

@@ -9,7 +9,7 @@
 require 'helpers/elasticsearch_options'
 require 'helpers/loggable_try'
 require 'logstash/ssl_file_tracker'
-require 'helpers/ssl_rebuildable'
+require 'helpers/elasticsearch_client_holder'
 require "license_checker/licensed"
 
 module LogStash
@@ -51,7 +51,7 @@ def initialize(settings, ssl_file_tracker = nil)
           ssl_file_tracker.register_paths(SSL_TRACK_ID_LICENSE, paths)
         end
 
-        @rebuildable = LogStash::Helpers::SslRebuildable.new(ssl_file_tracker, SSL_TRACK_ID_CPM) { build_client }
+        @es_client_holder = LogStash::Helpers::ElasticsearchClientHolder.create(ssl_file_tracker, SSL_TRACK_ID_CPM) { build_client }
         setup_license_checker(FEATURE_INTERNAL,
                               ssl_file_tracker: ssl_file_tracker,
                               tracking_id: SSL_TRACK_ID_LICENSE)
@@ -74,8 +74,6 @@ def get_pipeline_fetcher(es_version)
       def pipeline_configs
         logger.trace("Fetch remote config pipeline", :pipeline_ids => pipeline_ids)
 
-        @rebuildable.maybe_rebuild
-
         license_check(true)
         es_version = get_es_version
         fetcher = get_pipeline_fetcher(es_version)
@@ -196,7 +194,7 @@ def pipeline_ids
       end
 
       def client
-        @rebuildable.client
+        @es_client_holder.get
       end
     end
 

x-pack/lib/helpers/elasticsearch_client_holder.rb

@@ -0,0 +1,125 @@
+# Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+# or more contributor license agreements. Licensed under the Elastic License;
+# you may not use this file except in compliance with the Elastic License.
+
+require "logstash/logging/logger"
+
+module LogStash module Helpers
+  ##
+  # Holds an Elasticsearch client.
+  # @see ElasticsearchClientHolder#create
+  module ElasticsearchClientHolder
+
+    ##
+    # Creates an ElasticsearchClientHolder.
+    # If a `tracker` is specified, an SslRebuildable instance
+    # connected to the tracker with the provided id is returned;
+    # Otherwise, a Lazy instance is returned.
+    # @param tracker [#consume_stale, nil]
+    # @param id [#to_sym, nil]
+    # @yieldreturn [LogStash::Outputs::ElasticSearch::HttpClient]
+    # @return [ElasticsearchClientHolder]
+    def self.create(tracker=nil, id=nil, &client_factory)
+      return Lazy.new(&client_factory) if tracker.nil?
+
+      SslRebuildable.new(tracker, id, &client_factory)
+    end
+
+    ##
+    # Get a current client for immediate use.
+    # Consumers MUST NOT cache the returned client.
+    # @return [LogStash::Outputs::ElasticSearch::HttpClient]
+    def get
+      fail NotImplementedError
+    end
+
+    ##
+    # close the current client, if it exists and is connected
+    # @return [void]
+    def close
+      fail NotImplementedError
+    end
+
+    ##
+    # An ElasticsearchClientHolder that lazily creates the client when it is needed,
+    # caching the result indefinitely.
+    # @api internal (see ElasticsearchClientHolder::create)
+    class Lazy
+      include ElasticsearchClientHolder
+      include LogStash::Util::Loggable
+
+      ##
+      # @yieldreturn [LogStash::Outputs::ElasticSearch::HttpClient]
+      def initialize(&client_factory)
+        fail ArgumentError, "client_factory block is required" unless block_given?
+        @client_factory = client_factory
+      end
+
+      def get
+        @client || Util.synchronize(self) do
+          @client ||= begin
+                        logger.debug("initializing ES client")
+                        @client_factory.call
+                      end
+        end
+      end
+
+      def close
+        Util.synchronize(self) do
+          @client&.close
+        end
+      end
+    end
+
+    ##
+    # An ElasticsearchClientHolder that is connected to the provided tracker by the provided id.
+    # The client is created lazily, and is re-created when the tracker has marked the given id as stale.
+    # @api internal (see ElasticsearchClientHolder::create)
+    class SslRebuildable
+      include ElasticsearchClientHolder
+      include LogStash::Util::Loggable
+
+      attr_reader :tracker
+      attr_reader :id
+
+      ##
+      # @param tracker [#consume_stale]
+      # @param id [#to_sym]
+      # @yieldreturn [LogStash::Outputs::ElasticSearch::HttpClient]
+      def initialize(tracker, id, &client_factory)
+        @tracker = tracker or fail(ArgumentError, "tracker is required")
+        @id = id&.to_sym or fail(ArgumentError, "id is required")
+        @client_factory = client_factory or fail(ArgumentError, "client_factory block is required")
+      end
+
+      def get
+        Util.synchronize(self) do
+          @client ||= begin
+                        logger.debug("initializing rebuildable elasticsearch client `#{@id}`")
+                        @client_factory.call
+                      end
+
+          @tracker.consume_stale(@id) do
+            begin
+              old_client = @client
+              @client = @client_factory.call
+              logger.debug("rebuilt elasticsearch client `#{@id}` on certificate change")
+              old_client&.close rescue logger.warn("error closing stale elasticsearch client `#{@id}`", exception: $!.class, message: $!.message)
+            rescue => e
+              logger.warn("failed to rebuild elasticsearch client `#{@id}`", exception: e.class, message: e.message)
+              raise
+            end
+          end
+
+          @client
+        end
+      end
+
+      def close
+        Util.synchronize(self) do
+          @client&.close
+        end
+      end
+    end
+  end
+end end

x-pack/lib/license_checker/license_manager.rb

@@ -51,7 +51,6 @@ def serverless?
       end
 
       def fetch_license
-        @license_reader.maybe_rebuild_client
         fetch_cluster_info
         if serverless?
           update_xpack_info XPackInfo.serverless_response

x-pack/lib/license_checker/license_reader.rb

@@ -4,7 +4,7 @@
 
 require 'logstash/logging/logger'
 require 'logstash/outputs/elasticsearch'
-require 'helpers/ssl_rebuildable'
+require 'helpers/elasticsearch_client_holder'
 
 module LogStash
   module LicenseChecker
@@ -20,7 +20,7 @@ def initialize(settings, feature, options, ssl_file_tracker: nil, tracking_id: n
         es_options = options.merge('resurrect_delay' => 30)
         @es_options = Helpers::ElasticsearchOptions::es_options_with_product_origin_header(es_options)
 
-        @rebuildable = LogStash::Helpers::SslRebuildable.new(ssl_file_tracker, tracking_id) { build_client }
+        @es_client_holder = LogStash::Helpers::ElasticsearchClientHolder.create(ssl_file_tracker, tracking_id) { build_client }
       end
 
       ##
@@ -68,11 +68,7 @@ def fetch_cluster_info
       ##
       # @api private
       def client
-        @rebuildable.client
-      end
-
-      def maybe_rebuild_client
-        @rebuildable.maybe_rebuild
+        @es_client_holder.get
       end
 
       private

x-pack/spec/config_management/elasticsearch_source_spec.rb

@@ -873,7 +873,7 @@
 
       old_client = double("old_client")
       expect(old_client).to receive(:close)
-      instance.instance_variable_get(:@rebuildable).instance_variable_set(:@client, old_client)
+      instance.instance_variable_get(:@es_client_holder).instance_variable_set(:@client, old_client)
       allow(instance).to receive(:build_client).and_return(double("new_client"))
 
       instance.pipeline_configs
@@ -888,7 +888,7 @@
 
       old_client = double("old_client")
       expect(old_client).not_to receive(:close)
-      instance.instance_variable_get(:@rebuildable).instance_variable_set(:@client, old_client)
+      instance.instance_variable_get(:@es_client_holder).instance_variable_set(:@client, old_client)
 
       instance.pipeline_configs
     end

x-pack/spec/helpers/elasticsearch_client_holder_spec.rb

@@ -0,0 +1,104 @@
+# Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+# or more contributor license agreements. Licensed under the Elastic License;
+# you may not use this file except in compliance with the Elastic License.
+
+require "spec_helper"
+require "helpers/elasticsearch_client_holder"
+require "logstash/ssl_file_tracker"
+
+describe LogStash::Helpers::ElasticsearchClientHolder do
+  let(:tracker) { instance_double(LogStash::SslFileTracker) }
+  let(:tracking_id) { :".cpm" }
+  let(:factory_calls) { [] }
+  let(:factory) { ->() { factory_calls << :built; double("client", close: nil) } }
+  subject(:es_client_holder) { described_class.create(tracker, tracking_id, &factory) }
+
+  describe "#initialize" do
+    it "raises when no factory block is given" do
+      expect { described_class.create(tracker, :".cpm") }
+        .to raise_error(ArgumentError, /client_factory block is required/)
+    end
+  end
+
+  describe "::create" do
+    shared_examples "without factory block" do
+      it "fails helpfully when no factory block is provided" do
+        expect { described_class.create(tracker, tracking_id) }
+          .to raise_error(ArgumentError, /client_factory block is required/)
+      end
+    end
+
+    context "when no tracker is given" do
+      let(:tracker) { nil }
+      it "returns a Lazy instance" do
+        client_holder = described_class.create(tracker, tracking_id, &factory)
+        expect(client_holder).to be_a_kind_of LogStash::Helpers::ElasticsearchClientHolder
+        expect(client_holder).to be_a_kind_of LogStash::Helpers::ElasticsearchClientHolder::Lazy
+      end
+      include_examples "without factory block"
+    end
+
+    context "when a tracker and id are both given" do
+      it "returns an SslRebuildable instance with the provided tracker and id" do
+        client_holder = described_class.create(tracker, tracking_id, &factory)
+        expect(client_holder).to be_a_kind_of LogStash::Helpers::ElasticsearchClientHolder
+        expect(client_holder).to be_a_kind_of LogStash::Helpers::ElasticsearchClientHolder::SslRebuildable
+        expect(client_holder.tracker).to eq(tracker)
+        expect(client_holder.id).to eq(tracking_id)
+      end
+
+      include_examples "without factory block"
+    end
+
+    context "when `tracker` is provided and `id` is not" do
+      let(:tracking_id) { nil }
+      it "fails helpfully" do
+        expect { described_class.create(tracker, tracking_id) }.to raise_error(ArgumentError, /id is required/)
+      end
+    end
+  end
+
+  shared_examples "#get memoization" do
+    it "lazily calls the factory once and caches the result" do
+      es_client_holder
+      expect(factory_calls.size).to eq(0)
+      first = es_client_holder.get
+      second = es_client_holder.get
+      expect(factory_calls.size).to eq(1)
+      expect(second).to be(first)
+    end
+  end
+
+  describe "Lazy implementation" do
+    let(:es_client_holder) { described_class::Lazy.new(&factory) }
+
+    describe "#get" do
+      include_examples "#get memoization"
+    end
+  end
+
+  describe "SslRebuildable implementation" do
+    let(:es_client_holder) { described_class::SslRebuildable.new(tracker, tracking_id, &factory) }
+
+    describe "#get" do
+      before(:each) do
+        # mimic tracking behaviour when the id has NOT been marked stale
+        allow(tracker).to receive(:consume_stale).with(tracking_id).and_return(nil)
+      end
+      include_examples "#get memoization"
+
+      it "reloads the client when it has been invalidated" do
+        existing = es_client_holder.get
+        expect(existing).to receive(:close)
+
+        # mimic tracker behaviour when the id has been marked stale
+        allow(tracker).to receive(:consume_stale).with(tracking_id).and_yield
+
+        expect do
+          fresh = es_client_holder.get
+          expect(fresh).to_not be(existing)
+        end.to change { factory_calls.size }.by(1)
+      end
+    end
+  end
+end