Route badly interpolated dynamic index to DLQ (#1084)

* Added unit test to verify that bad replaced dynamic index name doesn't create an Elasticsearch index

* Added scaffolding to do not send to ES index dynamic names that are not fully resolved

* Fixed test to check no index is created in case of missed placeholder resolution in the index name

* Minor, removed commented code

* Route unresolved index name events to DLQ

* Extracted common dlq sending method

* Minor, removed commented code

* Added feature flag to eventually disable the DLQ-ing of not replaced indexnames

* Bumped version

Author: andsel

CHANGELOG.md

@@ -1,3 +1,6 @@
+## 11.9.0
+ - Feature: force unresolved dynamic index names to be sent into DLQ. This feature could be explicitly disabled using `dlq_on_failed_indexname_interpolation` setting [#1084](https://github.com/logstash-plugins/logstash-output-elasticsearch/pull/1084)
+
 ## 11.8.0
  - Feature: Adds a new `dlq_custom_codes` option to customize DLQ codes [#1067](https://github.com/logstash-plugins/logstash-output-elasticsearch/pull/1067)
 

docs/index.asciidoc

@@ -320,6 +320,7 @@ This plugin supports the following configuration options plus the
 | <<plugins-{type}s-{plugin}-data_stream_sync_fields>> |<<boolean,boolean>>|No
 | <<plugins-{type}s-{plugin}-data_stream_type>> |<<string,string>>|No
 | <<plugins-{type}s-{plugin}-dlq_custom_codes>> |<<number,number>>|No
+| <<plugins-{type}s-{plugin}-dlq_on_failed_indexname_interpolation>> |<<boolean,boolean>>|No
 | <<plugins-{type}s-{plugin}-doc_as_upsert>> |<<boolean,boolean>>|No
 | <<plugins-{type}s-{plugin}-document_id>> |<<string,string>>|No
 | <<plugins-{type}s-{plugin}-document_type>> |<<string,string>>|No
@@ -533,6 +534,14 @@ This list is an addition to the ordinary error codes considered for this feature
 It's considered a configuration error to re-use the same predefined codes for success, DLQ or conflict.
 The option accepts a list of natural numbers corresponding to HTTP errors codes.
 
+[id="plugins-{type}s-{plugin}-dlq_on_failed_indexname_interpolation"]
+===== `dlq_on_failed_indexname_interpolation`
+
+* Value type is <<boolean,boolean>>
+* Default value is `true`.
+
+If enabled, failed index name interpolation events go into dead letter queue.
+
 [id="plugins-{type}s-{plugin}-doc_as_upsert"]
 ===== `doc_as_upsert`
 

lib/logstash/outputs/elasticsearch.rb

@@ -261,6 +261,9 @@ class LogStash::Outputs::ElasticSearch < LogStash::Outputs::Base
   # The option accepts a list of natural numbers corresponding to HTTP errors codes.
   config :dlq_custom_codes, :validate => :number, :list => true, :default => []
 
+  # if enabled, failed index name interpolation events go into dead letter queue.
+  config :dlq_on_failed_indexname_interpolation, :validate => :boolean, :default => true
+
   attr_reader :client
   attr_reader :default_index
   attr_reader :default_ilm_rollover_alias
@@ -362,11 +365,43 @@ def config_init(params)
   # Receive an array of events and immediately attempt to index them (no buffering)
   def multi_receive(events)
     wait_for_successful_connection if @after_successful_connection_done
-    retrying_submit map_events(events)
+    events_mapped = safe_interpolation_map_events(events)
+    retrying_submit(events_mapped.successful_events)
+    unless events_mapped.failed_events.empty?
+      @logger.error("Can't map some events, needs to be handled by DLQ #{events_mapped.failed_events}")
+      send_failed_resolutions_to_dlq(events_mapped.failed_events)
+    end
+  end
+
+  # @param: Arrays of EventActionTuple
+  private
+  def send_failed_resolutions_to_dlq(failed_action_tuples)
+    failed_action_tuples.each do |action|
+      handle_dlq_status(action, "warn", "Could not resolve dynamic index")
+    end
+  end
+
+  MapEventsResult = Struct.new(:successful_events, :failed_events)
+
+  private
+  def safe_interpolation_map_events(events)
+    successful_events = [] # list of LogStash::Outputs::ElasticSearch::EventActionTuple
+    failed_events = [] # list of LogStash::Event
+    events.each do |event|
+       begin
+        successful_events << @event_mapper.call(event)
+       rescue IndexInterpolationError, e
+         action = event.sprintf(@action || 'index')
+         event_action_tuple = EventActionTuple.new(action, [], event)
+         failed_events << event_action_tuple
+       end
+    end
+    MapEventsResult.new(successful_events, failed_events)
   end
 
+  public
   def map_events(events)
-    events.map(&@event_mapper)
+    safe_interpolation_map_events(events).successful_events
   end
 
   def wait_for_successful_connection
@@ -441,12 +476,24 @@ def initialize(action, params, event, event_data = nil)
 
   end
 
+  class IndexInterpolationError < ArgumentError
+    attr_reader :bad_formatted_index
+
+    def initialize(bad_formatted_index)
+      super("Badly formatted index, after interpolation still contains placeholder: [#{bad_formatted_index}]")
+
+      @bad_formatted_index = bad_formatted_index
+    end
+  end
+
   # @return Hash (initial) parameters for given event
   # @private shared event params factory between index and data_stream mode
   def common_event_params(event)
+    sprintf_index = @event_target.call(event)
+    raise IndexInterpolationError, sprintf_index if sprintf_index.match(/%{.*?}/) && dlq_on_failed_indexname_interpolation
     params = {
         :_id => @document_id ? event.sprintf(@document_id) : nil,
-        :_index => @event_target.call(event),
+        :_index => sprintf_index,
         routing_field_name => @routing ? event.sprintf(@routing) : nil
     }
 

lib/logstash/plugin_mixins/elasticsearch/common.rb

@@ -206,19 +206,23 @@ def next_sleep_interval(current_interval)
       doubled > @retry_max_interval ? @retry_max_interval : doubled
     end
 
-    def handle_dlq_status(message, action, status, response)
+    def handle_dlq_response(message, action, status, response)
+      _, action_params = action.event, [action[0], action[1], action[2]]
+      
+      # TODO: Change this to send a map with { :status => status, :action => action } in the future
+      detailed_message = "#{message} status: #{status}, action: #{action_params}, response: #{response}"
+      
+      log_level = dig_value(response, 'index', 'error', 'type') == 'invalid_index_name_exception' ? :error : :warn
+      
+      handle_dlq_status(action, log_level, detailed_message)
+    end
+
+    def handle_dlq_status(action, log_level, message)
       # To support bwc, we check if DLQ exists. otherwise we log and drop event (previous behavior)
       if @dlq_writer
-        event, action = action.event, [action[0], action[1], action[2]]
-        # TODO: Change this to send a map with { :status => status, :action => action } in the future
-        @dlq_writer.write(event, "#{message} status: #{status}, action: #{action}, response: #{response}")
+        @dlq_writer.write(action.event, "#{message}")
       else
-        if dig_value(response, 'index', 'error', 'type') == 'invalid_index_name_exception'
-          level = :error
-        else
-          level = :warn
-        end
-        @logger.send level, message, status: status, action: action, response: response
+        @logger.send log_level, message
       end
     end
 
@@ -269,7 +273,7 @@ def submit(actions)
           @logger.warn "Failed action", status: status, action: action, response: response if log_failure_type?(error)
           next
         elsif @dlq_codes.include?(status)
-          handle_dlq_status("Could not index event to Elasticsearch.", action, status, response)
+          handle_dlq_response("Could not index event to Elasticsearch.", action, status, response)
           @document_level_metrics.increment(:non_retryable_failures)
           next
         else

logstash-output-elasticsearch.gemspec

@@ -1,6 +1,6 @@
 Gem::Specification.new do |s|
   s.name            = 'logstash-output-elasticsearch'
-  s.version         = '11.8.0'
+  s.version         = '11.9.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/index_spec.rb

@@ -1,5 +1,6 @@
 require_relative "../../../spec/es_spec_helper"
 require "logstash/outputs/elasticsearch"
+require 'cgi'
 
 describe "TARGET_BULK_BYTES", :integration => true do
   let(:target_bulk_bytes) { LogStash::Outputs::ElasticSearch::TARGET_BULK_BYTES }
@@ -45,16 +46,57 @@
   end
 end
 
-describe "indexing" do
+def curl_and_get_json_response(url, method: :get, retrieve_err_payload: false); require 'open3'
+  cmd = "curl -s -v --show-error #{curl_opts} -X #{method.to_s.upcase} -k #{url}"
+  begin
+    out, err, status = Open3.capture3(cmd)
+  rescue Errno::ENOENT
+    fail "curl not available, make sure curl binary is installed and available on $PATH"
+  end
+
+  if status.success?
+    http_status = err.match(/< HTTP\/1.1 (\d+)/)[1] || '0' # < HTTP/1.1 200 OK\r\n
+
+    if http_status.strip[0].to_i > 2
+      error = (LogStash::Json.load(out)['error']) rescue nil
+      if error
+        if retrieve_err_payload
+          return error
+        else
+          fail "#{cmd.inspect} received an error: #{http_status}\n\n#{error.inspect}"
+        end
+      else
+        warn out
+        fail "#{cmd.inspect} unexpected response: #{http_status}\n\n#{err}"
+      end
+    end
+
+    LogStash::Json.load(out)
+  else
+    warn out
+    fail "#{cmd.inspect} process failed: #{status}\n\n#{err}"
+  end
+end
+
+describe "indexing with sprintf resolution", :integration => true do
   let(:message) { "Hello from #{__FILE__}" }
   let(:event) { LogStash::Event.new("message" => message, "type" => type) }
-  let(:index) { 10.times.collect { rand(10).to_s }.join("") }
+  let (:index) { "%{[index_name]}_dynamic" }
   let(:type) { ESHelper.es_version_satisfies?("< 7") ? "doc" : "_doc" }
-  let(:event_count) { 1 + rand(2) }
-  let(:config) { "not implemented" }
+  let(:event_count) { 1 }
+  let(:user) { "simpleuser" }
+  let(:password) { "abc123" }
+  let(:config) do
+    {
+      "hosts" => [ get_host_port ],
+      "user" => user,
+      "password" => password,
+      "index" => index
+    }
+  end
   let(:events) { event_count.times.map { event }.to_a }
   subject { LogStash::Outputs::ElasticSearch.new(config) }
-  
+
   let(:es_url) { "http://#{get_host_port}" }
   let(:index_url) { "#{es_url}/#{index}" }
 
@@ -63,33 +105,65 @@
   let(:es_admin) { 'admin' } # default user added in ES -> 8.x requires auth credentials for /_refresh etc
   let(:es_admin_pass) { 'elastic' }
 
-  def curl_and_get_json_response(url, method: :get); require 'open3'
-    cmd = "curl -s -v --show-error #{curl_opts} -X #{method.to_s.upcase} -k #{url}"
-    begin
-      out, err, status = Open3.capture3(cmd)
-    rescue Errno::ENOENT
-      fail "curl not available, make sure curl binary is installed and available on $PATH"
-    end
+  let(:initial_events) { [] }
 
-    if status.success?
-      http_status = err.match(/< HTTP\/1.1 (\d+)/)[1] || '0' # < HTTP/1.1 200 OK\r\n
+  let(:do_register) { true }
 
-      if http_status.strip[0].to_i > 2
-        error = (LogStash::Json.load(out)['error']) rescue nil
-        if error
-          fail "#{cmd.inspect} received an error: #{http_status}\n\n#{error.inspect}"
-        else
-          warn out
-          fail "#{cmd.inspect} unexpected response: #{http_status}\n\n#{err}"
-        end
-      end
+  before do
+    subject.register if do_register
+    subject.multi_receive(initial_events) if initial_events
+  end
 
-      LogStash::Json.load(out)
-    else
-      warn out
-      fail "#{cmd.inspect} process failed: #{status}\n\n#{err}"
+  after do
+    subject.do_close
+  end
+
+  let(:event) { LogStash::Event.new("message" => message, "type" => type, "index_name" => "test") }
+
+  it "should index successfully when field is resolved" do
+    expected_index_name = "test_dynamic"
+    subject.multi_receive(events)
+
+#     curl_and_get_json_response "#{es_url}/_refresh", method: :post
+
+    result = curl_and_get_json_response "#{es_url}/#{expected_index_name}"
+
+    expect(result[expected_index_name]).not_to be(nil)
+  end
+
+  context "when dynamic field doesn't resolve the index_name" do
+    let(:event) { LogStash::Event.new("message" => message, "type" => type) }
+    let(:dlq_writer) { double('DLQ writer') }
+    before { subject.instance_variable_set('@dlq_writer', dlq_writer) }
+
+    it "should doesn't create an index name with unresolved placeholders" do
+      expect(dlq_writer).to receive(:write).once.with(event, /Could not resolve dynamic index/)
+      subject.multi_receive(events)
+
+      escaped_index_name = CGI.escape("%{[index_name]}_dynamic")
+      result = curl_and_get_json_response "#{es_url}/#{escaped_index_name}", retrieve_err_payload: true
+      expect(result["root_cause"].first()["type"]).to eq("index_not_found_exception")
     end
   end
+end
+
+describe "indexing" do
+  let(:message) { "Hello from #{__FILE__}" }
+  let(:event) { LogStash::Event.new("message" => message, "type" => type) }
+  let(:index) { 10.times.collect { rand(10).to_s }.join("") }
+  let(:type) { ESHelper.es_version_satisfies?("< 7") ? "doc" : "_doc" }
+  let(:event_count) { 1 + rand(2) }
+  let(:config) { "not implemented" }
+  let(:events) { event_count.times.map { event }.to_a }
+  subject { LogStash::Outputs::ElasticSearch.new(config) }
+  
+  let(:es_url) { "http://#{get_host_port}" }
+  let(:index_url) { "#{es_url}/#{index}" }
+
+  let(:curl_opts) { nil }
+
+  let(:es_admin) { 'admin' } # default user added in ES -> 8.x requires auth credentials for /_refresh etc
+  let(:es_admin_pass) { 'elastic' }
 
   let(:initial_events) { [] }
 

spec/unit/outputs/elasticsearch_spec.rb

@@ -768,6 +768,7 @@
 
   context 'handling elasticsearch document-level status meant for the DLQ' do
     let(:options) { { "manage_template" => false } }
+    let(:action) { LogStash::Outputs::ElasticSearch::EventActionTuple.new(:action, :params, LogStash::Event.new("foo" => "bar")) }
 
     context 'when @dlq_writer is nil' do
       before { subject.instance_variable_set '@dlq_writer', nil }
@@ -777,31 +778,28 @@
           it 'should log at ERROR level' do
             subject.instance_variable_set(:@logger, double("logger").as_null_object)
             mock_response = { 'index' => { 'error' => { 'type' => 'invalid_index_name_exception' } } }
-            subject.handle_dlq_status("Could not index event to Elasticsearch.",
-              [:action, :params, :event], :some_status, mock_response)
+            subject.handle_dlq_response("Could not index event to Elasticsearch.", action, :some_status, mock_response)
           end
         end
 
         context 'when getting any other exception' do
           it 'should log at WARN level' do
             logger = double("logger").as_null_object
             subject.instance_variable_set(:@logger, logger)
-            expect(logger).to receive(:warn).with(/Could not index/, hash_including(:status, :action, :response))
+            expect(logger).to receive(:warn).with(a_string_including "Could not index event to Elasticsearch. status: some_status, action: [:action, :params, {")
             mock_response = { 'index' => { 'error' => { 'type' => 'illegal_argument_exception' } } }
-            subject.handle_dlq_status("Could not index event to Elasticsearch.",
-              [:action, :params, :event], :some_status, mock_response)
+            subject.handle_dlq_response("Could not index event to Elasticsearch.", action, :some_status, mock_response)
           end
         end
 
         context 'when the response does not include [error]' do
           it 'should not fail, but just log a warning' do
             logger = double("logger").as_null_object
             subject.instance_variable_set(:@logger, logger)
-            expect(logger).to receive(:warn).with(/Could not index/, hash_including(:status, :action, :response))
+            expect(logger).to receive(:warn).with(a_string_including "Could not index event to Elasticsearch. status: some_status, action: [:action, :params, {")
             mock_response = { 'index' => {} }
             expect do
-              subject.handle_dlq_status("Could not index event to Elasticsearch.",
-                [:action, :params, :event], :some_status, mock_response)
+              subject.handle_dlq_response("Could not index event to Elasticsearch.", action, :some_status, mock_response)
             end.to_not raise_error
           end
         end
@@ -821,7 +819,7 @@
         expect(dlq_writer).to receive(:write).once.with(event, /Could not index/)
         mock_response = { 'index' => { 'error' => { 'type' => 'illegal_argument_exception' } } }
         action = LogStash::Outputs::ElasticSearch::EventActionTuple.new(:action, :params, event)
-        subject.handle_dlq_status("Could not index event to Elasticsearch.", action, 404, mock_response)
+        subject.handle_dlq_response("Could not index event to Elasticsearch.", action, 404, mock_response)
       end
     end