move discover_cluster_id into common

Author: colinsurprenant

lib/logstash/outputs/elasticsearch.rb

@@ -262,12 +262,18 @@ def register
     # To support BWC, we check if DLQ exists in core (< 5.4). If it doesn't, we use nil to resort to previous behavior.
     @dlq_writer = dlq_enabled? ? execution_context.dlq_writer : nil
 
+    check_action_validity
+
     # the license_checking behaviour in the Pool class is externalized in the LogStash::ElasticSearchOutputLicenseChecker
     # class defined in license_check.rb. This license checking is specific to the elasticsearch output here and passed
     # to build_client down to the Pool class.
     build_client(LicenseChecker.new(@logger))
-    setup_after_successful_connection
-    check_action_validity
+
+    @template_installer = setup_after_successful_connection do
+      discover_cluster_uuid
+      install_template
+      setup_ilm if ilm_in_use?
+    end
     @bulk_request_metrics = metric.namespace(:bulk_requests)
     @document_level_metrics = metric.namespace(:documents)
     @logger.info("New Elasticsearch output", :class => self.class.name, :hosts => @hosts.map(&:sanitized).map(&:to_s))
@@ -373,19 +379,6 @@ def retry_on_conflict_action_name
 
   private
 
-  def discover_cluster_uuid
-    return unless defined?(plugin_metadata)
-    cluster_info = client.get('/')
-    plugin_metadata.set(:cluster_uuid, cluster_info['cluster_uuid'])
-  rescue => e
-    # TODO introducing this logging message breaks many tests that need refactoring
-    # @logger.error("Unable to retrieve elasticsearch cluster uuid", error => e.message)
-  end
-
-  def successful_connection?
-    !!maximum_seen_major_version
-  end
-
   def routing_field_name
     maximum_seen_major_version >= 6 ? :routing : :_routing
   end
@@ -432,22 +425,6 @@ def install_template
     @template_installed.make_true
   end
 
-  def setup_after_successful_connection
-    @template_installer ||= Thread.new do
-      sleep_interval = @retry_initial_interval
-      until successful_connection? || @stopping.true?
-        @logger.debug("Waiting for connectivity to Elasticsearch cluster. Retrying in #{sleep_interval}s")
-        Stud.stoppable_sleep(sleep_interval) { @stopping.true? }
-        sleep_interval = next_sleep_interval(sleep_interval)
-      end
-      if successful_connection?
-        discover_cluster_uuid
-        install_template
-        setup_ilm if ilm_in_use?
-      end
-    end
-  end
-
   def setup_ecs_compatibility_related_defaults
     case ecs_compatibility
     when :disabled

lib/logstash/plugin_mixins/elasticsearch/common.rb

@@ -119,6 +119,34 @@ def maximum_seen_major_version
       client.maximum_seen_major_version
     end
 
+    def successful_connection?
+      !!maximum_seen_major_version
+    end
+
+    # launch a thread that waits for an initial successful connection to the ES cluster to call the given block
+    # @param block [Proc] the block to execute upon initial successful connection
+    # @return [Thread] the successful connection wait thread
+    def setup_after_successful_connection(&block)
+      Thread.new do
+        sleep_interval = @retry_initial_interval
+        until successful_connection? || @stopping.true?
+          @logger.debug("Waiting for connectivity to Elasticsearch cluster. Retrying in #{sleep_interval}s")
+          Stud.stoppable_sleep(sleep_interval) { @stopping.true? }
+          sleep_interval = next_sleep_interval(sleep_interval)
+        end
+        block.call if successful_connection?
+      end
+    end
+
+    def discover_cluster_uuid
+      return unless defined?(plugin_metadata)
+      cluster_info = client.get('/')
+      plugin_metadata.set(:cluster_uuid, cluster_info['cluster_uuid'])
+    rescue => e
+      # TODO introducing this logging message breaks many tests that need refactoring
+      # @logger.error("Unable to retrieve elasticsearch cluster uuid", error => e.message)
+    end
+
     def retrying_submit(actions)
       # Initially we submit the full list of actions
       submit_actions = actions