lib/fluent/plugin/out_elasticsearch.rb
# send_bulk given a specific bulk request, the original tag,
# chunk, and bulk_message_count
send_bulk given a specific bulk request, the original tag,
chunk, and bulk_message_count
send_bulk given a specific bulk request the original tag chunk and bulk_message_count
def send_bulk(data, tag, chunk, bulk_message_count, extracted_values, index)
      begin
        log.on_trace { log.trace "bulk request: #{data}" }
        response = client.bulk body: data, index: index
        log.on_trace { log.trace "bulk response: #{response}" }
        if response['errors']
          error = Fluent::Plugin::ElasticsearchErrorHandler.new(self)
          error.handle_error(response, tag, chunk, bulk_message_count, extracted_values)
        end
      rescue RetryStreamError => e
        emit_tag = @retry_tag ? @retry_tag : tag
        router.emit_stream(emit_tag, e.retry_stream)
      rescue => e
        @_es = nil if @reconnect_on_error
        @_es_info = nil if @reconnect_on_error
        raise RecoverableRequestFailure, "could not push logs to Elasticsearch cluster (#{connection_options_description}): #{e.message}"
      end
    end
def send_bulk ( data , tag , chunk , bulk_message_count , extracted_values , index ) begin log . on_trace { log . trace "bulk request: #{data}" } response = client . bulk body : data , index : index log . on_trace { log . trace "bulk response: #{response}" } if response [ 'errors' ] error = Fluent :: Plugin :: ElasticsearchErrorHandler . new ( self ) error . handle_error ( response , tag , chunk , bulk_message_count , extracted_values ) end rescue RetryStreamError => e emit_tag = @retry_tag ? @retry_tag : tag router . emit_stream ( emit_tag , e . retry_stream ) rescue => e @_es = nil if @reconnect_on_error @_es_info = nil if @reconnect_on_error raise RecoverableRequestFailure , "could not push logs to Elasticsearch cluster (#{connection_options_description}): #{e.message}" end end
send_bulk given a specific bulk request, the original tag,
chunk, and bulk_message_count
send_bulk given a specific bulk request the original tag chunk and bulk_message_count
# FIXME: identify unrecoverable errors and raise UnrecoverableRequestFailure instead
data
<not_specific>
tag
<not_specific>
chunk
<not_specific>
bulk_message_count
<not_specific>
extracted_values
<not_specific>
index
<not_specific>