separate common methods into elasticsearch specific and shared methods (logstash-plugins#976)

specific elasticsearch methods are now into the main elasticsearch
class file and shared methods have been moved into the
PluginMixins::ElasticSearch::Common namespace.
The license checking code has been externalized and can now be
specified as argument to the build_client method.
This is code refactorting that has no end-user impact.

Author: colinsurprenant

CHANGELOG.md

@@ -1,5 +1,6 @@
 ## unreleased
  - Refactored configuration options into specific and shared in PluginMixins namespace [#973](https://github.com/logstash-plugins/logstash-output-elasticsearch/pull/973)
+ - Refactored common methods into specific and shared in PluginMixins namespace [#976](https://github.com/logstash-plugins/logstash-output-elasticsearch/pull/976)
 
 ## 10.7.3
  - Added composable index template support for elasticsearch version 8 [#980](https://github.com/logstash-plugins/logstash-output-elasticsearch/pull/980)

lib/logstash/outputs/elasticsearch.rb

@@ -86,15 +86,16 @@
 class LogStash::Outputs::ElasticSearch < LogStash::Outputs::Base
   declare_threadsafe!
 
+  require "logstash/outputs/elasticsearch/license_checker"
   require "logstash/outputs/elasticsearch/http_client"
   require "logstash/outputs/elasticsearch/http_client_builder"
   require "logstash/plugin_mixins/elasticsearch/api_configs"
-  require "logstash/outputs/elasticsearch/common"
+  require "logstash/plugin_mixins/elasticsearch/common"
   require "logstash/outputs/elasticsearch/ilm"
   require 'logstash/plugin_mixins/ecs_compatibility_support'
 
   # Protocol agnostic methods
-  include(LogStash::Outputs::ElasticSearch::Common)
+  include(LogStash::PluginMixins::ElasticSearch::Common)
 
   # Methods for ILM support
   include(LogStash::Outputs::ElasticSearch::Ilm)
@@ -246,33 +247,37 @@ class LogStash::Outputs::ElasticSearch < LogStash::Outputs::Base
   # ILM policy to use, if undefined the default policy will be used.
   config :ilm_policy, :validate => :string, :default => DEFAULT_POLICY
 
+  attr_reader :default_index
+  attr_reader :default_ilm_rollover_alias
+  attr_reader :default_template_name
+
   def initialize(*params)
     super
     setup_ecs_compatibility_related_defaults
   end
 
-  def setup_ecs_compatibility_related_defaults
-    case ecs_compatibility
-    when :disabled
-      @default_index = "logstash-%{+yyyy.MM.dd}"
-      @default_ilm_rollover_alias = "logstash"
-      @default_template_name = 'logstash'
-    when :v1
-      @default_index = "ecs-logstash-%{+yyyy.MM.dd}"
-      @default_ilm_rollover_alias = "ecs-logstash"
-      @default_template_name = 'ecs-logstash'
-    else
-      fail("unsupported ECS Compatibility `#{ecs_compatibility}`")
-    end
+  def register
+    @template_installed = Concurrent::AtomicBoolean.new(false)
+    @stopping = Concurrent::AtomicBoolean.new(false)
+    # 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
 
-    @index ||= default_index
-    @ilm_rollover_alias ||= default_ilm_rollover_alias
-    @template_name ||= default_template_name
-  end
+    check_action_validity
 
-  attr_reader :default_index
-  attr_reader :default_ilm_rollover_alias
-  attr_reader :default_template_name
+    # 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))
+
+    @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))
+  end
 
   # @override to handle proxy => '' as if none was set
   def config_init(params)
@@ -290,20 +295,12 @@ def config_init(params)
     super(params)
   end
 
-  def build_client
-    # the following 3 options validation & setup methods are called inside build_client
-    # because they must be executed prior to building the client and logstash
-    # monitoring and management rely on directly calling build_client
-    # see https://github.com/logstash-plugins/logstash-output-elasticsearch/pull/934#pullrequestreview-396203307
-    validate_authentication
-    fill_hosts_from_cloud_id
-    setup_hosts
-
-    params["metric"] = metric
-    if @proxy.eql?('')
-      @logger.warn "Supplied proxy setting (proxy => '') has no effect"
+  # Receive an array of events and immediately attempt to index them (no buffering)
+  def multi_receive(events)
+    until @template_installed.true?
+      sleep 1
     end
-    @client ||= ::LogStash::Outputs::ElasticSearch::HttpClientBuilder.build(@logger, @hosts, params)
+    retrying_submit(events.map {|e| event_action_tuple(e)})
   end
 
   def close
@@ -312,8 +309,65 @@ def close
     @client.close if @client
   end
 
-  def self.oss?
-    LogStash::OSS
+  # not private because used by ILM specs
+  def stop_template_installer
+    @template_installer.join unless @template_installer.nil?
+  end
+
+  # not private for elasticsearch_spec.rb
+  # Convert the event into a 3-tuple of action, params, and event
+  def event_action_tuple(event)
+    action = event.sprintf(@action)
+
+    params = {
+      :_id => @document_id ? event.sprintf(@document_id) : nil,
+      :_index => event.sprintf(@index),
+      routing_field_name => @routing ? event.sprintf(@routing) : nil
+    }
+
+    params[:_type] = get_event_type(event) if use_event_type?(nil)
+
+    if @pipeline
+      value = event.sprintf(@pipeline)
+      # convention: empty string equates to not using a pipeline
+      # this is useful when using a field reference in the pipeline setting, e.g.
+      #      elasticsearch {
+      #        pipeline => "%{[@metadata][pipeline]}"
+      #      }
+      params[:pipeline] = value unless value.empty?
+    end
+
+    if @parent
+      if @join_field
+        join_value = event.get(@join_field)
+        parent_value = event.sprintf(@parent)
+        event.set(@join_field, { "name" => join_value, "parent" => parent_value })
+        params[routing_field_name] = event.sprintf(@parent)
+      else
+        params[:parent] = event.sprintf(@parent)
+      end
+    end
+
+    if action == 'update'
+      params[:_upsert] = LogStash::Json.load(event.sprintf(@upsert)) if @upsert != ""
+      params[:_script] = event.sprintf(@script) if @script != ""
+      params[retry_on_conflict_action_name] = @retry_on_conflict
+    end
+
+    if @version
+      params[:version] = event.sprintf(@version)
+    end
+
+    if @version_type
+      params[:version_type] = event.sprintf(@version_type)
+    end
+
+    [action, params, event]
+  end
+
+  # not private for elasticsearch_spec.rb
+  def retry_on_conflict_action_name
+    maximum_seen_major_version >= 7 ? :retry_on_conflict : :_retry_on_conflict
   end
 
   @@plugins = Gem::Specification.find_all{|spec| spec.name =~ /logstash-output-elasticsearch-/ }
@@ -322,4 +376,87 @@ def self.oss?
     name = plugin.name.split('-')[-1]
     require "logstash/outputs/elasticsearch/#{name}"
   end
+
+  private
+
+  def routing_field_name
+    maximum_seen_major_version >= 6 ? :routing : :_routing
+  end
+
+  # Determine the correct value for the 'type' field for the given event
+  DEFAULT_EVENT_TYPE_ES6="doc".freeze
+  DEFAULT_EVENT_TYPE_ES7="_doc".freeze
+  def get_event_type(event)
+    # Set the 'type' value for the index.
+    type = if @document_type
+             event.sprintf(@document_type)
+           else
+             if maximum_seen_major_version < 6
+               event.get("type") || DEFAULT_EVENT_TYPE_ES6
+             elsif maximum_seen_major_version == 6
+               DEFAULT_EVENT_TYPE_ES6
+             elsif maximum_seen_major_version == 7
+               DEFAULT_EVENT_TYPE_ES7
+             else
+               nil
+             end
+           end
+
+    if !(type.is_a?(String) || type.is_a?(Numeric))
+      @logger.warn("Bad event type! Non-string/integer type value set!", :type_class => type.class, :type_value => type.to_s, :event => event)
+    end
+
+    type.to_s
+  end
+
+  ##
+  # WARNING: This method is overridden in a subclass in Logstash Core 7.7-7.8's monitoring,
+  #          where a `client` argument is both required and ignored. In later versions of
+  #          Logstash Core it is optional and ignored, but to make it optional here would
+  #          allow us to accidentally break compatibility with Logstashes where it was required.
+  # @param noop_required_client [nil]: required `nil` for legacy reasons.
+  # @return [Boolean]
+  def use_event_type?(noop_required_client)
+    maximum_seen_major_version < 8
+  end
+
+  def install_template
+    TemplateManager.install_template(self)
+    @template_installed.make_true
+  end
+
+  def setup_ecs_compatibility_related_defaults
+    case ecs_compatibility
+    when :disabled
+      @default_index = "logstash-%{+yyyy.MM.dd}"
+      @default_ilm_rollover_alias = "logstash"
+      @default_template_name = 'logstash'
+    when :v1
+      @default_index = "ecs-logstash-%{+yyyy.MM.dd}"
+      @default_ilm_rollover_alias = "ecs-logstash"
+      @default_template_name = 'ecs-logstash'
+    else
+      fail("unsupported ECS Compatibility `#{ecs_compatibility}`")
+    end
+
+    @index ||= default_index
+    @ilm_rollover_alias ||= default_ilm_rollover_alias
+    @template_name ||= default_template_name
+  end
+
+  # To be overidden by the -java version
+  VALID_HTTP_ACTIONS=["index", "delete", "create", "update"]
+  def valid_actions
+    VALID_HTTP_ACTIONS
+  end
+
+  def check_action_validity
+    raise LogStash::ConfigurationError, "No action specified!" unless @action
+
+    # If we're using string interpolation, we're good!
+    return if @action =~ /%{.+}/
+    return if valid_actions.include?(@action)
+
+    raise LogStash::ConfigurationError, "Action '#{@action}' is invalid! Pick one of #{valid_actions} or use a sprintf style statement"
+  end
 end

lib/logstash/outputs/elasticsearch/http_client.rb

@@ -286,6 +286,7 @@ def build_pool(options)
       adapter = build_adapter(options)
 
       pool_options = {
+        :license_checker => options[:license_checker],
         :sniffing => sniffing,
         :sniffer_delay => options[:sniffer_delay],
         :sniffing_path => options[:sniffing_path],

lib/logstash/outputs/elasticsearch/http_client/pool.rb

@@ -1,3 +1,5 @@
+require "logstash/plugin_mixins/elasticsearch/noop_license_checker"
+
 module LogStash; module Outputs; class ElasticSearch; class HttpClient;
   class Pool
     class NoConnectionAvailableError < Error; end
@@ -29,6 +31,7 @@ def message
     end
 
     attr_reader :logger, :adapter, :sniffing, :sniffer_delay, :resurrect_delay, :healthcheck_path, :sniffing_path, :bulk_path
+    attr_reader :license_checker # license_checker is used by the pool specs
 
     ROOT_URI_PATH = '/'.freeze
     LICENSE_PATH = '/_license'.freeze
@@ -66,12 +69,10 @@ def initialize(logger, adapter, initial_urls=[], options={})
       # Holds metadata about all URLs
       @url_info = {}
       @stopping = false
-    end
 
-    def oss?
-     LogStash::Outputs::ElasticSearch.oss?
+      @license_checker = options.fetch(:license_checker) { LogStash::PluginMixins::ElasticSearch::NoopLicenseChecker::INSTANCE }
     end
-    
+
     def start
       update_initial_urls
       start_resurrectionist
@@ -210,7 +211,6 @@ def address_str_to_uri(addr_str)
       end
     end
 
-
     def sniff_2x_1x(nodes)
       nodes.map do |id,info|
         # TODO Make sure this works with shield. Does that listed
@@ -245,16 +245,15 @@ def start_resurrectionist
       end
     end
 
+    # Retrieve ES node license information
+    # @param url [LogStash::Util::SafeURI] ES node URL
+    # @return [Hash] deserialized license document or empty Hash upon any error
     def get_license(url)
       response = perform_request_to_url(url, :get, LICENSE_PATH)
       LogStash::Json.load(response.body)
-    end
-
-    def valid_es_license?(url)
-      license = get_license(url)
-      license.fetch("license", {}).fetch("status", nil) == "active"
     rescue => e
-      false
+      logger.error("Unable to get license information", url: url.sanitized.to_s, error_type: e.class, error: e.message)
+      {}
     end
 
     def health_check_request(url)
@@ -282,19 +281,9 @@ def healthcheck!
               @logger.warn("Detected a node with a higher major version than previously observed. This could be the result of an elasticsearch cluster upgrade.", :previous_major => @maximum_seen_major_version, :new_major => major, :node_url => url.sanitized.to_s)
               set_new_major_version(major)
             end
-            if oss? || valid_es_license?(url)
-              meta[:state] = :alive
-            else
-              # As this version is to be shipped with Logstash 7.x we won't mark the connection as unlicensed
-              #
-              #  logger.error("Cannot connect to the Elasticsearch cluster configured in the Elasticsearch output. Logstash requires the default distribution of Elasticsearch. Please update to the default distribution of Elasticsearch for full access to all free features, or switch to the OSS distribution of Logstash.", :url => url.sanitized.to_s)
-              #  meta[:state] = :unlicensed
-              #
-              # Instead we'll log a deprecation warning and mark it as alive:
-              #
-              log_license_deprecation_warn(url)
-              meta[:state] = :alive
-            end
+
+            alive = @license_checker.appropriate_license?(self, url)
+            meta[:state] = alive ? :alive : :dead
           end
         rescue HostUnreachableError, BadResponseCodeError => e
           logger.warn("Attempted to resurrect connection to dead ES instance, but got an error.", url: url.sanitized.to_s, error_type: e.class, error: e.message)
@@ -306,10 +295,6 @@ def stop_resurrectionist
       @resurrectionist.join if @resurrectionist
     end
 
-    def log_license_deprecation_warn(url)
-      logger.warn("DEPRECATION WARNING: Connecting to an OSS distribution of Elasticsearch using the default distribution of Logstash will stop working in Logstash 8.0.0. Please upgrade to the default distribution of Elasticsearch, or use the OSS distribution of Logstash", :url => url.sanitized.to_s)
-    end
-
     def resurrectionist_alive?
       @resurrectionist ? @resurrectionist.alive? : nil
     end

lib/logstash/outputs/elasticsearch/http_client_builder.rb

@@ -15,6 +15,7 @@ def self.build(logger, hosts, params)
       client_settings[:proxy] = params["proxy"] if params["proxy"]
 
       common_options = {
+        :license_checker => params["license_checker"],
         :client_settings => client_settings,
         :metric => params["metric"],
         :resurrect_delay => params["resurrect_delay"]