lib/logstash/plugin_mixins/elasticsearch/common.rb
def safe_bulk(actions)
      sleep_interval = @retry_initial_interval
      begin
        @client.bulk(actions) # returns { 'errors': ..., 'items': ... }
      rescue ::LogStash::Outputs::ElasticSearch::HttpClient::Pool::HostUnreachableError => e
        # If we can't even connect to the server let's just print out the URL (:hosts is actually a URL)
        # and let the user sort it out from there
        @logger.error(
          "Attempted to send a bulk request but Elasticsearch appears to be unreachable or down",
          message: e.message, exception: e.class, will_retry_in_seconds: sleep_interval
        )
        @logger.debug? && @logger.debug("Failed actions for last bad bulk request", :actions => actions)

        # We retry until there are no errors! Errors should all go to the retry queue
        sleep_interval = sleep_for_interval(sleep_interval)
        @bulk_request_metrics.increment(:failures)
        retry unless @stopping.true?
      rescue ::LogStash::Outputs::ElasticSearch::HttpClient::Pool::NoConnectionAvailableError => e
        @logger.error(
          "Attempted to send a bulk request but there are no living connections in the pool " +
          "(perhaps Elasticsearch is unreachable or down?)",
          message: e.message, exception: e.class, will_retry_in_seconds: sleep_interval
        )

        sleep_interval = sleep_for_interval(sleep_interval)
        @bulk_request_metrics.increment(:failures)
        retry unless @stopping.true?
      rescue ::LogStash::Outputs::ElasticSearch::HttpClient::Pool::BadResponseCodeError => e
        @bulk_request_metrics.increment(:failures)
        log_hash = {:code => e.response_code, :url => e.url.sanitized.to_s, :content_length => e.request_body.bytesize}
        log_hash[:body] = e.response_body if @logger.debug? # Generally this is too verbose
        message = "Encountered a retryable error (will retry with exponential backoff)"

        # We treat 429s as a special case because these really aren't errors, but
        # rather just ES telling us to back off a bit, which we do.
        # The other retryable code is 503, which are true errors
        # Even though we retry the user should be made aware of these
        if e.response_code == 429
          logger.debug(message, log_hash)
        else
          logger.error(message, log_hash)
        end

        sleep_interval = sleep_for_interval(sleep_interval)
        retry
      rescue => e # Stuff that should never happen - print out full connection issues
        @logger.error(
          "An unknown error occurred sending a bulk request to Elasticsearch (will retry indefinitely)",
          message: e.message, exception: e.class, backtrace: e.backtrace
        )
        @logger.debug? && @logger.debug("Failed actions for last bad bulk request", :actions => actions)

        sleep_interval = sleep_for_interval(sleep_interval)
        @bulk_request_metrics.increment(:failures)
        retry unless @stopping.true?
      end
    end
# Rescue retryable errors during bulk submission
# @param actions a [action, params, event.to_hash] tuple
# @return response [Hash] which contains 'errors' and processed 'items' entries
Rescue retryable errors during bulk submission
Rescue retryable errors during bulk submission
def safe_bulk(actions)
      sleep_interval = @retry_initial_interval
      begin
        @client.bulk(actions) 
      rescue ::LogStash::Outputs::ElasticSearch::HttpClient::Pool::HostUnreachableError => e
        @logger.error(
          "Attempted to send a bulk request but Elasticsearch appears to be unreachable or down",
          message: e.message, exception: e.class, will_retry_in_seconds: sleep_interval
        )
        @logger.debug? && @logger.debug("Failed actions for last bad bulk request", :actions => actions)
        sleep_interval = sleep_for_interval(sleep_interval)
        @bulk_request_metrics.increment(:failures)
        retry unless @stopping.true?
      rescue ::LogStash::Outputs::ElasticSearch::HttpClient::Pool::NoConnectionAvailableError => e
        @logger.error(
          "Attempted to send a bulk request but there are no living connections in the pool " +
          "(perhaps Elasticsearch is unreachable or down?)",
          message: e.message, exception: e.class, will_retry_in_seconds: sleep_interval
        )
        sleep_interval = sleep_for_interval(sleep_interval)
        @bulk_request_metrics.increment(:failures)
        retry unless @stopping.true?
      rescue ::LogStash::Outputs::ElasticSearch::HttpClient::Pool::BadResponseCodeError => e
        @bulk_request_metrics.increment(:failures)
        log_hash = {:code => e.response_code, :url => e.url.sanitized.to_s, :content_length => e.request_body.bytesize}
        log_hash[:body] = e.response_body if @logger.debug? 
        message = "Encountered a retryable error (will retry with exponential backoff)"
        if e.response_code == 429
          logger.debug(message, log_hash)
        else
          logger.error(message, log_hash)
        end
        sleep_interval = sleep_for_interval(sleep_interval)
        retry
      rescue => e 
        @logger.error(
          "An unknown error occurred sending a bulk request to Elasticsearch (will retry indefinitely)",
          message: e.message, exception: e.class, backtrace: e.backtrace
        )
        @logger.debug? && @logger.debug("Failed actions for last bad bulk request", :actions => actions)
        sleep_interval = sleep_for_interval(sleep_interval)
        @bulk_request_metrics.increment(:failures)
        retry unless @stopping.true?
      end
    end
def safe_bulk ( actions ) sleep_interval = @retry_initial_interval begin @client . bulk ( actions ) rescue :: LogStash :: Outputs :: ElasticSearch :: HttpClient :: Pool :: HostUnreachableError => e @logger . error ( "Attempted to send a bulk request but Elasticsearch appears to be unreachable or down" , message : e . message , exception : e . class , will_retry_in_seconds : sleep_interval ) @logger . debug? && @logger . debug ( "Failed actions for last bad bulk request" , :actions => actions ) sleep_interval = sleep_for_interval ( sleep_interval ) @bulk_request_metrics . increment ( :failures ) retry unless @stopping . true? rescue :: LogStash :: Outputs :: ElasticSearch :: HttpClient :: Pool :: NoConnectionAvailableError => e @logger . error ( "Attempted to send a bulk request but there are no living connections in the pool " + "(perhaps Elasticsearch is unreachable or down?)" , message : e . message , exception : e . class , will_retry_in_seconds : sleep_interval ) sleep_interval = sleep_for_interval ( sleep_interval ) @bulk_request_metrics . increment ( :failures ) retry unless @stopping . true? rescue :: LogStash :: Outputs :: ElasticSearch :: HttpClient :: Pool :: BadResponseCodeError => e @bulk_request_metrics . increment ( :failures ) log_hash = { :code => e . response_code , :url => e . url . sanitized . to_s , :content_length => e . request_body . bytesize } log_hash [ :body ] = e . response_body if @logger . debug? message = "Encountered a retryable error (will retry with exponential backoff)" if e . response_code == 429 logger . debug ( message , log_hash ) else logger . error ( message , log_hash ) end sleep_interval = sleep_for_interval ( sleep_interval ) retry rescue => e @logger . error ( "An unknown error occurred sending a bulk request to Elasticsearch (will retry indefinitely)" , message : e . message , exception : e . class , backtrace : e . backtrace ) @logger . debug? && @logger . debug ( "Failed actions for last bad bulk request" , :actions => actions ) sleep_interval = sleep_for_interval ( sleep_interval ) @bulk_request_metrics . increment ( :failures ) retry unless @stopping . true? end end
Rescue retryable errors during bulk submission
Rescue retryable errors during bulk submission
# returns { 'errors': ..., 'items': ... } # If we can't even connect to the server let's just print out the URL (:hosts is actually a URL) # and let the user sort it out from there # We retry until there are no errors! Errors should all go to the retry queue # Generally this is too verbose # We treat 429s as a special case because these really aren't errors, but # rather just ES telling us to back off a bit, which we do. # The other retryable code is 503, which are true errors # Even though we retry the user should be made aware of these # Stuff that should never happen - print out full connection issues
response  which contains 'errors' and processed 'items' entries
response which contains ' errors ' and processed ' items ' entries
Hash
actions
None