Re-shape the logic to avoid unnecessary usage of Tuple, instead plain Event is considered. Giving a noti to users when DQL enabled and failed events sent to it.

Author: mashhurs

Diff for: lib/logstash/outputs/elasticsearch.rb

@@ -372,29 +372,33 @@ def multi_receive(events)
     end
   end
 
-  # @param: Arrays of EventActionTuple with error messages
+  # @param: Arrays of FailedEventMapping
   private
   def handle_event_mapping_errors(event_mapping_errors)
-    event_mapping_errors.each do |action, message|
-      detail_message = message + ", " + action.to_s
-      handle_dlq_status(action.event, :warn, detail_message)
-      @document_level_metrics.increment(:non_retryable_failures) unless @dlq_writer
+    # if DQL is enabled, log the events to provide issue insights to users.
+    if @dlq_writer
+      @logger.warn("Events could not be indexed and routing to DLQ, count: #{event_mapping_errors.size}")
     end
+
+    event_mapping_errors.each do |event_mapping_error|
+      detailed_message = "#{event_mapping_error.message}; event: `#{event_mapping_error.event.to_hash_with_metadata}`"
+      handle_dlq_status(event_mapping_error.event, :warn, detailed_message)
+    end
+    @document_level_metrics.increment(:non_retryable_failures, event_mapping_errors.size)
   end
 
   MapEventsResult = Struct.new(:successful_events, :event_mapping_errors)
+  FailedEventMapping = Struct.new(:event, :message)
 
   private
   def safe_interpolation_map_events(events)
     successful_events = [] # list of LogStash::Outputs::ElasticSearch::EventActionTuple
-    event_mapping_errors = [] # list of LogStash::Outputs::ElasticSearch::EventActionTuple with error messages
+    event_mapping_errors = [] # list of FailedEventMapping
     events.each do |event|
       begin
         successful_events << @event_mapper.call(event)
       rescue EventMappingError => ie
-        action = event.sprintf(@action || 'index')
-        event_action_tuple = EventActionTuple.new(action, [], event)
-        event_mapping_errors << [event_action_tuple, ie.message]
+        event_mapping_errors << FailedEventMapping.new(event, ie.message)
        end
     end
     MapEventsResult.new(successful_events, event_mapping_errors)

Diff for: spec/integration/outputs/index_spec.rb

@@ -137,7 +137,7 @@ def curl_and_get_json_response(url, method: :get, retrieve_err_payload: false);
     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/)
+      expect(dlq_writer).to receive(:write).once.with(event, a_string_including("Badly formatted index, after interpolation still contains placeholder"))
       subject.multi_receive(events)
 
       escaped_index_name = CGI.escape("%{[index_name]}_dynamic")

Diff for: spec/unit/outputs/elasticsearch_spec.rb

@@ -398,6 +398,9 @@
           event_result.successful_events.each do |action, _|
             expect(action).to_not eql("unsupported_action")
           end
+          event_result.event_mapping_errors.each do |event_mapping_error|
+            expect(event_mapping_error.message).to eql("Elasticsearch doesn't support [unsupported_action] action")
+          end
         end
       end
 
@@ -414,6 +417,9 @@
           event_result.successful_events.each do |action, _|
             expect(action).to_not eql("unsupported_action")
           end
+          event_result.event_mapping_errors.each do |event_mapping_error|
+            expect(event_mapping_error.message).to eql("Elasticsearch doesn't support [unsupported_action] action")
+          end
         end
       end
 
@@ -430,6 +436,9 @@
           event_result.successful_events.each do |action, _|
             expect(action).to_not eql("unsupported_action")
           end
+          event_result.event_mapping_errors.each do |event_mapping_error|
+            expect(event_mapping_error.message).to include "Elasticsearch doesn't support"
+          end
         end
       end
 
@@ -443,7 +452,10 @@
         it "rejects unsupported actions" do
           event_result = subject.send(:safe_interpolation_map_events, events)
           expect(event_result.successful_events.size).to be == 2
-          # expect(logger_stub).to have_received(:warn).with(a_string_including "Could not index event to Elasticsearch because its action is not supported.")
+          expect(event_result.event_mapping_errors.size).to be == 2
+          event_result.event_mapping_errors.each do |event_mapping_error|
+            expect(event_mapping_error.message).to include "Elasticsearch doesn't support"
+          end
         end
       end
 
@@ -457,7 +469,9 @@
         it "rejects unsupported action" do
           event_result = subject.send(:safe_interpolation_map_events, events)
           expect(event_result.successful_events.size).to be == 3
-          # expect(logger_stub).to have_received(:warn).with(a_string_including "Could not index event to Elasticsearch because its action is not supported.")
+          event_result.event_mapping_errors.each do |event_mapping_error|
+            expect(event_mapping_error.message).to eql("Elasticsearch doesn't support [unsupported_action3] action")
+          end
         end
       end
     end