lib/logstash/plugin_mixins/elasticsearch/common.rb
# This module defines common methods that can be reused by alternate elasticsearch output plugins such as the elasticsearch_data_streams output.
This module defines common methods that can be reused by alternate elasticsearch output plugins such as the elasticsearch_data_streams output.
This module defines common methods that can be reused by alternate elasticsearch output plugins such as the elasticsearch_data_streams output .
def submit(actions)
      bulk_response = safe_bulk(actions)
      return if bulk_response.nil?
      if bulk_response["errors"]
        @bulk_request_metrics.increment(:with_errors)
      else
        @bulk_request_metrics.increment(:successes)
        @document_level_metrics.increment(:successes, actions.size)
        return
      end
      responses = bulk_response["items"]
      if responses.size != actions.size 
        msg = "Sent #{actions.size} documents but Elasticsearch returned #{responses.size} responses"
        @logger.warn(msg, actions: actions, responses: responses)
        fail("#{msg} (likely a bug with _bulk endpoint)")
      end
      actions_to_retry = []
      responses.each_with_index do |response,idx|
        action_type, action_props = response.first
        status = action_props["status"]
        error  = action_props["error"]
        action = actions[idx]
        action_params = action[1]
        if DOC_SUCCESS_CODES.include?(status)
          @document_level_metrics.increment(:successes)
          next
        elsif DOC_CONFLICT_CODE == status
          @document_level_metrics.increment(:non_retryable_failures)
          @logger.warn "Failed action", status: status, action: action, response: response if log_failure_type?(error)
          next
        elsif DOC_DLQ_CODES.include?(status)
          handle_dlq_status("Could not index event to Elasticsearch.", action, status, response)
          @document_level_metrics.increment(:non_retryable_failures)
          next
        else
          @document_level_metrics.increment(:retryable_failures)
          @logger.info "Retrying failed action", status: status, action: action, error: error if log_failure_type?(error)
          actions_to_retry << action
        end
      end
      actions_to_retry
    end
def submit ( actions ) bulk_response = safe_bulk ( actions ) return if bulk_response . nil? if bulk_response [ "errors" ] @bulk_request_metrics . increment ( :with_errors ) else @bulk_request_metrics . increment ( :successes ) @document_level_metrics . increment ( :successes , actions . size ) return end responses = bulk_response [ "items" ] if responses . size != actions . size msg = "Sent #{actions.size} documents but Elasticsearch returned #{responses.size} responses" @logger . warn ( msg , actions : actions , responses : responses ) fail ( "#{msg} (likely a bug with _bulk endpoint)" ) end actions_to_retry = [ ] responses . each_with_index do | response , idx | action_type , action_props = response . first status = action_props [ "status" ] error = action_props [ "error" ] action = actions [ idx ] action_params = action [ 1 ] if DOC_SUCCESS_CODES . include? ( status ) @document_level_metrics . increment ( :successes ) next elsif DOC_CONFLICT_CODE == status @document_level_metrics . increment ( :non_retryable_failures ) @logger . warn "Failed action" , status : status , action : action , response : response if log_failure_type? ( error ) next elsif DOC_DLQ_CODES . include? ( status ) handle_dlq_status ( "Could not index event to Elasticsearch." , action , status , response ) @document_level_metrics . increment ( :non_retryable_failures ) next else @document_level_metrics . increment ( :retryable_failures ) @logger . info "Retrying failed action" , status : status , action : action , error : error if log_failure_type? ( error ) actions_to_retry << action end end actions_to_retry end
This module defines common methods that can be reused by alternate elasticsearch output plugins such as the elasticsearch_data_streams output.
This module defines common methods that can be reused by alternate elasticsearch output plugins such as the elasticsearch_data_streams output .
# If the response is nil that means we were in a retry loop # and aborted since we're shutting down # If it did return and there are no errors we're good as well # can not map action -> response reliably # an ES bug (on 7.10.2, 7.11.1) where a _bulk request to index X documents would return Y (> X) items # Retry logic: If it is success, we move on. If it is a failure, we have 3 paths: # - For 409, we log and drop. there is nothing we can do # - For a mapping error, we send to dead letter queue for a human to intervene at a later point. # - For everything else there's mastercard. Yep, and we retry indefinitely. This should fix #572 and other transient network issues # only log what the user whitelisted
actions