shut down pipeline when finish register fails (#1151)

Prior to this change, the errors in the post-register phase (finish_register) behaved differently: template manager logged failure; ILM setup loop printed error messages. Continuing the pipeline would not be meaningful. With this commit, ConfigurationError is introduced for critical bootstrap failures, such as when the template path is not found. This change stops the pipeline when it gets ConfigurationError or 4xx status code calling template or ILM related API. For HTTP 429 too many request, finish_register retries the bootstrap --------- Co-authored-by: João Duarte <jsvd@users.noreply.github.com>

Author: kaisecheng

.travis.yml

@@ -2,10 +2,8 @@ import:
 - logstash-plugins/.ci:travis/travis.yml@1.x
 
 env:
- - INTEGRATION=false ELASTIC_STACK_VERSION=6.x
  - INTEGRATION=false ELASTIC_STACK_VERSION=7.x
  - INTEGRATION=false ELASTIC_STACK_VERSION=8.x SNAPSHOT=true
- - INTEGRATION=true ELASTIC_STACK_VERSION=6.x
  - INTEGRATION=true ELASTIC_STACK_VERSION=7.x
  - INTEGRATION=true ELASTIC_STACK_VERSION=7.x SNAPSHOT=true LOG_LEVEL=info
  - INTEGRATION=true ELASTIC_STACK_VERSION=8.x SNAPSHOT=true LOG_LEVEL=info

CHANGELOG.md

@@ -1,5 +1,8 @@
+## 11.20.0
+ - Changed the register to initiate pipeline shutdown upon bootstrap failure instead of simply logging the error [#1151](https://github.com/logstash-plugins/logstash-output-elasticsearch/pull/1151)
+
 ## 11.19.0
- - Added `filter_path` to bulk requests to reduce the size of responses from elasticsearch
+ - Added `filter_path` to bulk requests to reduce the size of responses from elasticsearch [#1154](https://github.com/logstash-plugins/logstash-output-elasticsearch/pull/1154)
 
 ## 11.18.0
  - Added request header `Elastic-Api-Version` for serverless [#1147](https://github.com/logstash-plugins/logstash-output-elasticsearch/pull/1147)

lib/logstash/outputs/elasticsearch.rb

@@ -322,12 +322,29 @@ def register
  @bulk_request_metrics = metric.namespace(:bulk_requests)
  @document_level_metrics = metric.namespace(:documents)
 
+ @shutdown_from_finish_register = Concurrent::AtomicBoolean.new(false)
  @after_successful_connection_thread = after_successful_connection do
  begin
  finish_register
  true # thread.value
+ rescue LogStash::ConfigurationError, LogStash::Outputs::ElasticSearch::HttpClient::Pool::BadResponseCodeError => e
+ return e if pipeline_shutdown_requested?
+
+ # retry when 429
+ @logger.debug("Received a 429 status code during registration. Retrying..") && retry if too_many_requests?(e)
+
+ # shut down pipeline
+ if execution_context&.agent.respond_to?(:stop_pipeline)
+ details = { message: e.message, exception: e.class }
+ details[:backtrace] = e.backtrace if @logger.debug?
+ @logger.error("Failed to bootstrap. Pipeline \"#{execution_context.pipeline_id}\" is going to shut down", details)
+
+ @shutdown_from_finish_register.make_true
+ execution_context.agent.stop_pipeline(execution_context.pipeline_id)
+ end
+
+ e
  rescue => e
- # we do not want to halt the thread with an exception as that has consequences for LS
  e # thread.value
  ensure
  @after_successful_connection_done.make_true
@@ -450,7 +467,11 @@ def close
  private
 
  def stop_after_successful_connection_thread
- @after_successful_connection_thread.join unless @after_successful_connection_thread.nil?
+ # avoid deadlock when finish_register calling execution_context.agent.stop_pipeline
+ # stop_pipeline triggers plugin close and the plugin close waits for after_successful_connection_thread to join
+ return if @shutdown_from_finish_register&.true?
+
+ @after_successful_connection_thread.join if @after_successful_connection_thread&.alive?
  end
 
  # Convert the event into a 3-tuple of action, params and event hash
@@ -599,6 +620,7 @@ def install_template
  details = { message: e.message, exception: e.class, backtrace: e.backtrace }
  details[:body] = e.response_body if e.respond_to?(:response_body)
  @logger.error("Failed to install template", details)
+ raise e if register_termination_error?(e)
  end
 
  def setup_ecs_compatibility_related_defaults

lib/logstash/outputs/elasticsearch/data_stream_support.rb

@@ -145,12 +145,12 @@ def inherited_internal_config_param?(name)
  # @note assumes to be running AFTER {after_successful_connection} completed, due ES version checks
  # @return [Gem::Version] if ES supports DS nil (or raise) otherwise
  def assert_es_version_supports_data_streams
- fail 'no last_es_version' unless last_es_version # assert - should not happen
+ raise LogStash::ConfigurationError 'no last_es_version' unless last_es_version # assert - should not happen
  es_version = ::Gem::Version.create(last_es_version)
  if es_version < ::Gem::Version.create(DATA_STREAMS_ORIGIN_ES_VERSION)
  @logger.error "Elasticsearch version does not support data streams, Logstash might end up writing to an index", es_version: es_version.version
  # NOTE: when switching to synchronous check from register, this should be a ConfigurationError
- raise LogStash::Error, "A data_stream configuration is only supported since Elasticsearch #{DATA_STREAMS_ORIGIN_ES_VERSION} " +
+ raise LogStash::ConfigurationError, "A data_stream configuration is only supported since Elasticsearch #{DATA_STREAMS_ORIGIN_ES_VERSION} " +
  "(detected version #{es_version.version}), please upgrade your cluster"
  end
  es_version # return truthy

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

@@ -28,7 +28,12 @@ def forbidden?
  @response_code == 403
  end
 
+ def too_many_requests?
+ @response_code == 429
+ end
+
  end
+
  class HostUnreachableError < Error;
  attr_reader :original_error, :url
 
@@ -69,7 +74,7 @@ def initialize(logger, adapter, initial_urls=[], options={})
  @adapter = adapter
  @metric = options[:metric]
  @initial_urls = initial_urls
- 
+
  raise ArgumentError, "No URL Normalizer specified!" unless options[:url_normalizer]
  @url_normalizer = options[:url_normalizer]
  DEFAULT_OPTIONS.merge(options).tap do |merged|
@@ -159,7 +164,7 @@ def until_stopped(task_name, delay)
  :error_message => e.message,
  :class => e.class.name,
  :backtrace => e.backtrace
-  )
+ )
  end
  end
  end
@@ -197,11 +202,11 @@ def check_sniff
  sniff(nodes)
  end
  end
- 
+
  def major_version(version_string)
  version_string.split('.').first.to_i
  end
- 
+
  def sniff(nodes)
  nodes.map do |id,info|
  # Skip master-only nodes
@@ -360,7 +365,7 @@ def normalize_url(uri)
 
  def update_urls(new_urls)
  return if new_urls.nil?
- 
+
  # Normalize URLs
  new_urls = new_urls.map(&method(:normalize_url))
 
@@ -388,14 +393,14 @@ def update_urls(new_urls)
  if state_changes[:removed].size > 0 || state_changes[:added].size > 0
  logger.info? && logger.info("Elasticsearch pool URLs updated", :changes => state_changes)
  end
- 
+
  # Run an inline healthcheck anytime URLs are updated
  # This guarantees that during startup / post-startup
  # sniffing we don't have idle periods waiting for the
  # periodic sniffer to allow new hosts to come online
- healthcheck! 
+ healthcheck!
  end
- 
+
  def size
  @state_mutex.synchronize { @url_info.size }
  end

lib/logstash/outputs/elasticsearch/template_manager.rb

@@ -37,7 +37,7 @@ def self.load_default_template(es_major_version, ecs_compatibility)
  template_path = default_template_path(es_major_version, ecs_compatibility)
  read_template_file(template_path)
  rescue => e
- fail "Failed to load default template for Elasticsearch v#{es_major_version} with ECS #{ecs_compatibility}; caused by: #{e.inspect}"
+ raise LogStash::ConfigurationError, "Failed to load default template for Elasticsearch v#{es_major_version} with ECS #{ecs_compatibility}; caused by: #{e.inspect}"
  end
 
  def self.install(client, template_endpoint, template_name, template, template_overwrite)
@@ -99,9 +99,11 @@ def self.default_template_path(es_major_version, ecs_compatibility=:disabled)
  end
 
  def self.read_template_file(template_path)
- raise ArgumentError, "Template file '#{template_path}' could not be found" unless ::File.exists?(template_path)
+ raise LogStash::ConfigurationError, "Template file '#{template_path}' could not be found" unless ::File.exists?(template_path)
  template_data = ::IO.read(template_path)
  LogStash::Json.load(template_data)
+ rescue => e
+ raise LogStash::ConfigurationError, "Failed to load template file '#{template_path}': #{e.message}"
  end
 
  def self.template_endpoint(plugin)

lib/logstash/plugin_mixins/elasticsearch/common.rb

@@ -407,5 +407,14 @@ def dig_value(val, first_key, *rest_keys)
  return val if rest_keys.empty? || val == nil
  dig_value(val, *rest_keys)
  end
+
+ def register_termination_error?(e)
+ e.is_a?(LogStash::ConfigurationError) || e.is_a?(LogStash::Outputs::ElasticSearch::HttpClient::Pool::BadResponseCodeError)
+ end
+
+ def too_many_requests?(e)
+ e.is_a?(LogStash::Outputs::ElasticSearch::HttpClient::Pool::BadResponseCodeError) &&
+ e.too_many_requests?
+ end
  end
 end; end; end

logstash-output-elasticsearch.gemspec

@@ -1,6 +1,6 @@
 Gem::Specification.new do |s|
  s.name = 'logstash-output-elasticsearch'
- s.version = '11.19.0'
+ s.version = '11.20.0'
  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/integration/outputs/no_es_on_startup_spec.rb

@@ -33,25 +33,25 @@
  end
 
  it 'should ingest events when Elasticsearch recovers before documents are sent' do
- allow_any_instance_of(LogStash::Outputs::ElasticSearch::HttpClient::Pool).to receive(:get_es_version).and_raise(
+ allow_any_instance_of(LogStash::Outputs::ElasticSearch::HttpClient::Pool).to receive(:get_root_path).with(any_args).and_raise(
  ::LogStash::Outputs::ElasticSearch::HttpClient::Pool::HostUnreachableError.new StandardError.new("TEST: before docs are sent"), 'http://test.es/'
  )
  subject.register
- allow_any_instance_of(LogStash::Outputs::ElasticSearch::HttpClient::Pool).to receive(:get_es_version).and_return(ESHelper.es_version)
+ allow_any_instance_of(LogStash::Outputs::ElasticSearch::HttpClient::Pool).to receive(:get_root_path).with(any_args).and_call_original
  subject.multi_receive([event1, event2])
  @es.indices.refresh
  r = @es.search(index: 'logstash-*')
  expect(r).to have_hits(2)
  end
 
  it 'should ingest events when Elasticsearch recovers after documents are sent' do
- allow_any_instance_of(LogStash::Outputs::ElasticSearch::HttpClient::Pool).to receive(:get_es_version).and_raise(
+ allow_any_instance_of(LogStash::Outputs::ElasticSearch::HttpClient::Pool).to receive(:get_root_path).with(any_args).and_raise(
  ::LogStash::Outputs::ElasticSearch::HttpClient::Pool::HostUnreachableError.new StandardError.new("TEST: after docs are sent"), 'http://test.es/'
  )
  subject.register
  Thread.new do
  sleep 4
- allow_any_instance_of(LogStash::Outputs::ElasticSearch::HttpClient::Pool).to receive(:get_es_version).and_return(ESHelper.es_version)
+ allow_any_instance_of(LogStash::Outputs::ElasticSearch::HttpClient::Pool).to receive(:get_root_path).with(any_args).and_call_original
  end
  subject.multi_receive([event1, event2])
  @es.indices.refresh