evaluate bulk request failures and reroute failed messages

[jcantrill]

instead of resubmitting the entire request

This PR add logic to evaluate a bulk request failure and to resubmit to the pipeline only those records that failed. Previously, all records in a bulk submission were resubmitted to Elastic even if there was only one failure in the entire request. This causes undue processing on both the collector and aggregator

(check all that apply)

• tests added
• tests passing
• README updated (if needed)
• README Table of Contents updated (if needed)
• History.md and version in gemspec are untouched
• backward compatible
• feature works in elasticsearch_dynamic (not required but recommended)

[jcantrill]

cc @richm @portante @cosmo0920

[richm]

lgtm

[portante]

s/succeede/succeed/, and missing trailing ")".

[portante]

Won't this use more memory on the non-error path as well, which we hope would be the fast path? Is there a way we can just pass the "chunk" object and only when we find some errors, we index into the chunk object to pull out the original records to re-submit?

In general, it seems that the non-error case should not have to shoulder any of the burden for error handling, and the error path becomes more costly to compensate.

[richm]

in fact, you have to use the raw record from the chunk - you need the resubmitted record to look exactly like the raw record from the chunk, so it can be reprocessed. So yes - we need the chunk, and in the error handing code we'll need to iterate the records using chunk.msgpack_each do |time, record|

[jcantrill]

Agree with your observation but Rich is correct, we are basing the functionality on the assumption that the order of records submitted is the same as the items in the response. This assumption is based upon Elasticsearch documentation as there is no way to otherwise correlate the request and response without ids or some hashing. This is also necessary to only keep a list of the records that can be properly deserialized

[portante]

But records ends up having the records in the same order as chunk.msgpack_each emits them, right? If that is the case, why do we need to construct a separate array on the normal path? Why not pass chunk in, and when we encounter errors, then construct the list to handle the errors?

[richm]

We have to iterate the chunk in the error handling. We need to make sure to get the same records that were sent, which means adding a couple of the record processing steps: omit records that are not Hash objects, and omit records that were sent to the dlq i.e. not included in the bulk index request. We can't do any more processing on the records - records resubmitted to fluentd must be the raw records directly from the queue.

[jcantrill]

Again, we cant iterate the chunk but the list of records that were actually submitted.

[richm]

need to use the chunk to get the raw, unprocessed record to resubmit

[portante]

But records ends up having the records in the same order as chunk.msgpack_each emits them, right? If that is the case, why do we need to construct a separate array on the normal path? Why not pass chunk in, and when we encounter errors, then construct the list to handle the errors?

[cosmo0920]

Sending bad record handling in ES plugin is not good manner. This behavior will make confusion.
I guess that this should be done with secondary output feature.

[jcantrill]

@cosmo0920 I don't understand what you are saying or your requested changes. This PR does nothing with 'bad record handling' that is not already in the plugin. It provides a way to reprocess a failed bulk request by:

• Evaluating the individual entries from the bulk
• 'accepting'/discarding successful or duplicates
• pushing failed entries back into the message pipeline for resubmission

We see errors in production right now where one failed insert from the bulk causes every message from the bulk insert to be attempted again adding undo load on Elasticsearch.

[cosmo0920]

@cosmo0920 I don't understand what you are saying or your requested changes. This PR does nothing with 'bad record handling' that is not already in the plugin. It provides a way to reprocess a failed bulk request by:

• Evaluating the individual entries from the bulk
• 'accepting'/discarding successful or duplicates
• pushing failed entries back into the message pipeline for resubmission

I don't have OpenShift environment, but I guess that if ES plugin can handle <secondary> option, failing sending records will be fallbacked into secondary output.
This fallback mechanism is not enough for your OpenShift environment?
Fluentd plugin can specify secondary output and ES plugin also can specify it:

% bundle exec fluent-plugin-config-format --format=txt output elasticsearch -p lib/fluent/plugin
helpers: event_emitter,compat_parameters,record_accessor
@log_level: string: (nil)
time_as_integer: bool: (false)
slow_flush_log_threshold: float: (20.0)
host: string: ("localhost")
port: integer: (9200)
user: string: (nil)
password: string: (nil)
path: string: (nil)
scheme: string: ("http")
hosts: string: (nil)
target_index_key: string: (nil)
target_type_key: string: (nil)
time_key_format: string: (nil)
time_precision: integer: (9)
include_timestamp: bool: (false)
logstash_format: bool: (false)
logstash_prefix: string: ("logstash")
logstash_prefix_separator: string: ("-")
logstash_dateformat: string: ("%Y.%m.%d")
utc_index: bool: (true)
type_name: string: ("fluentd")
index_name: string: ("fluentd")
id_key: string: (nil)
write_operation: string: ("index")
parent_key: string: (nil)
routing_key: string: (nil)
request_timeout: time: (5)
reload_connections: bool: (true)
reload_on_failure: bool: (false)
resurrect_after: time: (60)
time_key: string: (nil)
time_key_exclude_timestamp: bool: (false)
ssl_verify: bool: (true)
client_key: string: (nil)
client_cert: string: (nil)
client_key_pass: string: (nil)
ca_file: string: (nil)
ssl_version: enum: (:TLSv1)
remove_keys: string: (nil)
remove_keys_on_update: string: ("")
remove_keys_on_update_key: string: (nil)
flatten_hashes: bool: (false)
flatten_hashes_separator: string: ("_")
template_name: string: (nil)
template_file: string: (nil)
template_overwrite: bool: (false)
templates: hash: (nil)
include_tag_key: bool: (false)
tag_key: string: ("tag")
time_parse_error_tag: string: ("Fluent::ElasticsearchOutput::TimeParser.error")
reconnect_on_error: bool: (false)
pipeline: string: (nil)
with_transporter_log: bool: (false)
content_type: enum: (:"application/json")
<buffer>: optional, multiple
 @type: : ("memory")
 chunk_keys: : (["tag"])
 timekey_use_utc: : (true)
<secondary>: optional, single
 @type: string: (nil)
 <buffer>: optional, single
 <secondary>: optional, single

[jcantrill]

@cosmo0920 Im still not understanding what you are saying regarding a 'secondary' option. The flaw is between this plugin and fluentd, there is currently no way to partially accept the result of trying to submit a chunk. It is unrelated to openshift but a flaw in how this plugin sends to ES. If only one record fails, the exception bubbles up to fluentd and the entire chunk is re-tried. You can demonstrate this with any ES instance by restricting its memory and cpu and the pounding it with a huge message load. You should see failures due to memory exhaustion or the like. There is no need to push to a secondary output; the records just need to be sent back into the queue for processing which I believe is what this does.

[portante]

@cosmo0920, another way to think about this is that the Elasticsearch plugin is constructing a set of instructions for indexing the individual records of a chunk. Each record has its own success/failure status. That is not the case with other plugins, like secure_forward, where the entire chunk is transmitted as a whole, and the success/failure is of the entire chunk, not for each record in the chunk.

[jcantrill]

@cosmo0920 I'm going to try to setup a meeting with you but please consider this scenario which is the case we have seen with some of our configurations of Elasticsearch:

• ES memory is not sufficient to properly handle the message load
• ES frequently exhausts its memory allocated for bulk index requests
• ES CPU is not sufficent to process bulk indexing, querying, etc

The flow of fluentd with this output plugin is:

1. Read chunk
2. create bulk index request (many records)
3. POST to ES
4. Non-200 or 201 response like with several hundred records:

      {
      "took" : 1,
      "errors" : true,
      "items" : [
        {
          "create" : {
            "_index" : "foo",
            "_type"  : "bar",
            "_id" : "xyz",
            "status": 200
          },
          "create" : {
            "_index" : "foo",
            "_type"  : "bar",
            "_id" : "abc",
            "status": 500,
             "type": "out_of_memory_error",
             "reason" => "Java heap space"
          }
        }
      ]
     }

1. Raise exception from fluent-plugin-elasticsearch
2. Catch exception in fluentd
3. Repeat steps, reprocessing chunk while N < retry_limit

This current flow between fluent and this plugin has no logic to resubmit ONLY failed messages. You have to resubmit the entire request or non at all. It requires you to resubmit messages that were indexed before. This means resubmitting every successful record, every duplicate record, and every failed record. In our configuration this means 8Mb message chunks from up to 1000 or more instance of fluentd. This means EVERY record that already exists in Elasticsearch is being reserialized and retransmitted across the network because possibly one, single record in the bulk request failed.

Regarding secondary output:

At Buffered output plugin, the user can specify <secondary> with any output plugin in
 <match> configuration. If the retry count exceeds the buffer’s retry_limit (and the retry limit 
has not been disabled via disable_retry_limit), then buffered chunk is output to <secondary> output plugin.

<secondary> is useful for backup when destination servers are unavailable, e.g. forward, 
mongo and other plugins. We strongly recommend out_file plugin for <secondary>.

This means secondary_output is only used after the retry_limit is exceeded. I presume this also means you have no information about which records succeeded in order to determine which ones should be resubmitted. Additionally, since this is only after exceeding the retry_limit, you have executed the previously referenced logic many times over before even considering the secondary logic.

I believe the proposed change is well within what is reasonable based upon fluent's guidelines. Each output destination needs to be handled according to the characteristics of the target destination. In the case of Elasticsearch, this change addresses how to properly interact with bulk requests. Obviously, other output plugins may not need such handling thus there is no 'one size fits all'.

[cosmo0920]

but please consider this scenario which is the case we have seen with some of our configurations of Elasticsearch:

• ES memory is not sufficient to properly handle the message load
• ES frequently exhausts its memory allocated for bulk index requests
• ES CPU is not sufficent to process bulk indexing, querying, etc

Unfortunately, I don't enough knowledge about ES cluster operation.
Is there any room for consideration about buffer configuration?
It seems that Fluentd instance(s) send too large chunk to handle into ES cluster.

• Set smaller buffer_chunk_limit parameter
• Set adequate size of buffer_queue_limit parameter

can change each of chunk message size.

@cosmo0920, another way to think about this is that the Elasticsearch plugin is constructing a set of instructions for indexing the individual records of a chunk. Each record has its own success/failure status. That is not the case with other plugins, like secure_forward, where the entire chunk is transmitted as a whole, and the success/failure is of the entire chunk, not for each record in the chunk.

Should we provide one by one insert in ES plugin?
If so, it is very hard to implement in v0.12 branch.
With using v1 API, we can implement one by one insert and change behavior with configuration feature.

[richm]

please review https://www.elastic.co/guide/en/elasticsearch/guide/2.x/_monitoring_individual_nodes.html#_threadpool_section

We are trying to implement the following algorithm:

It is much better to handle queuing in your application by gracefully handling the back pressure from a full queue. When you receive bulk rejections, you should take these steps:

    Pause the import thread for 3–5 seconds.
    Extract the rejected actions from the bulk response, since it is probable that many of the actions were successful. The bulk response will tell you which succeeded and which were rejected.
    Send a new bulk request with just the rejected actions.
    Repeat from step 1 if rejections are encountered again. 

Using this procedure, your code naturally adapts to the load of your cluster and naturally backs off.

Rejections are not errors: they just mean you should try again later.

That is, we are trying to wait for a bit, then resubmit only the rejected actions.

[cosmo0920]

@richm Thank you for the information!
I want to know why this PR is needed.

It is reasonable to add this feature into master branch not v0.12 branch.

Because:

• Fluentd v0.12 is still old stable but moves maintenance phase
• ES plugin should follow Fluentd core development cycle

[richm]

Fluentd v0.12 is still old stable but moves maintenance phase

Ok, but this is a pretty critical bug - anyone running Elasticsearch in a non-trivial environment will run into this situation sooner or later, and there is no good way to mitigate this without the fluent-plugin-elasticsearch handling the backpressure correctly.

[jcantrill]

@cosmo0920 I rebased and fixed up the tests. Please reconsider merging this change as @richm stated this will affect anyone who has a non-trival environment. It's not properly solvable by tweaking buffer settings. We will otherwise need to carry a patch ourselves.

cc @portante please review to see if I have properly addressed your error handling concerns

[richm]

pass chunk instead of records

[jcantrill]

We can not pass chunk instead of records. This code is executed after we have deserialized the individual records in the chunk to confirm they dont throw exceptions like 'utf-8' problems. The chunk records are not representative of those we submitted to Elasticsearch

[richm]

also iterate chunk - pull the next item from chunk
You'll also have to implement the next unless record.is_a? Hash logic, and any logic in process_message that omits the record from the bulk index request. That will need to be refactored into a method callable from both places. You have to guarantee that the record you read from the chunk is the same one being processed in this error handling loop.

[jcantrill]

Same here. We've already processed the chunk. There is no reason to do so here too.

[richm]

this will be the record from the chunk instead

[richm]

this will be the record from the chunk instead

[richm]

this will be the record from the chunk instead

[cosmo0920]

@cosmo0920 I rebased and fixed up the tests. Please reconsider merging this change as @richm stated this will affect anyone who has a non-trival environment. It's not properly solvable by tweaking buffer settings. We will otherwise need to carry a patch ourselves.

Now, https://docs.fluentd.org/ points v1 document by default.
Fluentd now comers will start to use Fluentd v1 instead of v0.12.

I don't think that it is worth to add this back pressure feature into v0.12 branch from now.

[cosmo0920]

I decided to approve this PR because:

• Users cannot avoid flood of resubmitting entire chunk on current implementation.

[cosmo0920]

@jcantrill I replaced retry_records with retry_stream which is MultiEventStream instance.
I think that we can create an event stream for resubmitting within error handler.
What do you think?

[jcantrill]

Seems reasonable

[cosmo0920]

@jcantrill
When we create retry_stream which is MultiEventStream instance in error handler, we can pass created stream directly and become resubmitting code simple.
Is there any concern about this?

[jcantrill]

Only one which was brought up by @richm . He noted that when we resubmit records, I assume with the same tag, they go through the entire pipeline. This means if you have multiple filters or send messages to multiple outputs, you may be reprocessing(e.g. adding fields again, contacting other servers to get metadata) and duplicating message to other destinations (e.g. if we copied to say file). @richm and I talked about routing the messages in a similar way as @ERROR with providing a label like @RETRY or something. I recommend we may wish in this case to have a configurable label for records we wish to retry:

config_param: :retry_es_label, default: @RETRY_ES
...
router.emit_stream(@retry_es_label, e.retry_stream)
...

[richm]

I'm not sure how to do this in the code, but it was easy to add a @label @RETRY_ES to our es plugin config, then add a <label @RETRY_ES> section.

[cosmo0920]

Current Fluentd v0.12 (and v1) cannot specify label at runtime.
It is determined in configuration file at bootup phase.
Instead, we can reroute with adding prefix/suffix into tag like as:

config_param: :retry_es_tag_prefix, default: "retrying."
...
router.emit_stream(@retry_es_tag_prefix + tag, e.retry_stream)
...

[richm]

ok - so either the pipeline must be reentrant (that is, no label is required, it is ok for records to be resubmitted to the entire pipeline), or add some sort of @label @RETRY_ES to the fluent-plugin-elasticsearch config, with a <label @RETRY_ES> section in your fluent.conf, which guarantees that only records which should be retried for submission to elasticsearch go into this label processing.

[jcantrill]

@cosmo0920 Can we not assume that users will have to specify the label in their config pipeline and that the 'configuration' is simply them telling us which label to use. Will routing fail if the label does not exist at runtime?

[richm]

yes, routing will fail at startup - if there is a @label @FOO declaration without a <label @FOO>, or vice versa, in the fluent.conf

[cosmo0920]

@label is optional configuration element.
If there is no specified label, events are re-emitted into same route and cause infinite busy loop.

[cosmo0920]

Mismatched arguments between caller and callee.
Is this intentional?

[richm]

no, not intentional - good catch

[cosmo0920]

Test passing and I reviewed again.
It seems good to me!
I've merged this PR.
Thank you for your work, @jcantrill !

[jcantrill]

@cosmo0920 We'll also work on a subsequent PR to allow a 'retry' tag which would allow you to possibly reroute messages through a different pipeline. For the one we use, we desire to skip section of the pipeline and simply resubmit the message to skip specific filters.

[cosmo0920]

Yup. We should still work on retrying mechanism.

[deiwin]

This and the other emit_error_event usage below seem to be problematic. The router API expects the error to be an Exception. This is causing errors in the logs:

2018-05-08 14:45:51 +0000 [warn]: #0 failed to flush the buffer. retry_time=0 next_retry_seconds=2018-05-08 14:45:52 +0000 chunk="56bb2d75cc799064fe3005a7ce548c43" error_class=NoMethodError error="undefined method `backtrace' for \"400 - Rejected by Elasticsearch\":String"
  2018-05-08 14:45:51 +0000 [warn]: #0 /opt/td-agent/embedded/lib/ruby/gems/2.4.0/gems/fluentd-1.0.2/lib/fluent/root_agent.rb:304:in `emit_error_event'
  2018-05-08 14:45:51 +0000 [warn]: #0 /opt/td-agent/embedded/lib/ruby/gems/2.4.0/gems/fluentd-1.0.2/lib/fluent/event_router.rb:102:in `emit_error_event'
  2018-05-08 14:45:51 +0000 [warn]: #0 /opt/td-agent/embedded/lib/ruby/gems/2.4.0/gems/fluent-plugin-elasticsearch-2.10.0/lib/fluent/plugin/elasticsearch_error_handler.rb:64:in `block in handle_error'
  2018-05-08 14:45:51 +0000 [warn]: #0 /opt/td-agent/embedded/lib/ruby/gems/2.4.0/gems/fluentd-1.0.2/lib/fluent/event.rb:323:in `each'
  2018-05-08 14:45:51 +0000 [warn]: #0 /opt/td-agent/embedded/lib/ruby/gems/2.4.0/gems/fluentd-1.0.2/lib/fluent/event.rb:323:in `block in each'
  2018-05-08 14:45:51 +0000 [warn]: #0 /opt/td-agent/embedded/lib/ruby/gems/2.4.0/gems/fluentd-1.0.2/lib/fluent/plugin/buffer/memory_chunk.rb:80:in `open'
  2018-05-08 14:45:51 +0000 [warn]: #0 /opt/td-agent/embedded/lib/ruby/gems/2.4.0/gems/fluentd-1.0.2/lib/fluent/plugin/buffer/memory_chunk.rb:80:in `open'
  2018-05-08 14:45:51 +0000 [warn]: #0 /opt/td-agent/embedded/lib/ruby/gems/2.4.0/gems/fluentd-1.0.2/lib/fluent/event.rb:322:in `each'
  2018-05-08 14:45:51 +0000 [warn]: #0 /opt/td-agent/embedded/lib/ruby/gems/2.4.0/gems/fluent-plugin-elasticsearch-2.10.0/lib/fluent/plugin/elasticsearch_error_handler.rb:27:in `handle_error'
  2018-05-08 14:45:51 +0000 [warn]: #0 /opt/td-agent/embedded/lib/ruby/gems/2.4.0/gems/fluent-plugin-elasticsearch-2.10.0/lib/fluent/plugin/out_elasticsearch.rb:510:in `send_bulk'
  2018-05-08 14:45:51 +0000 [warn]: #0 /opt/td-agent/embedded/lib/ruby/gems/2.4.0/gems/fluent-plugin-elasticsearch-2.10.0/lib/fluent/plugin/out_elasticsearch.rb:408:in `write'
  2018-05-08 14:45:51 +0000 [warn]: #0 /opt/td-agent/embedded/lib/ruby/gems/2.4.0/gems/fluentd-1.0.2/lib/fluent/plugin/output.rb:1093:in `try_flush'
  2018-05-08 14:45:51 +0000 [warn]: #0 /opt/td-agent/embedded/lib/ruby/gems/2.4.0/gems/fluentd-1.0.2/lib/fluent/plugin/output.rb:1318:in `flush_thread_run'
  2018-05-08 14:45:51 +0000 [warn]: #0 /opt/td-agent/embedded/lib/ruby/gems/2.4.0/gems/fluentd-1.0.2/lib/fluent/plugin/output.rb:439:in `block (2 levels) in start'
  2018-05-08 14:45:51 +0000 [warn]: #0 /opt/td-agent/embedded/lib/ruby/gems/2.4.0/gems/fluentd-1.0.2/lib/fluent/plugin_helper/thread.rb:78:in `block in thread_create'