High CPU Usage with elasticsearch plugin when ES is not reachable (or API failed)

[williamzhao87]

Describe the bug
When ES is down, FluentD CPU climbs to 50-70% usage, due to fluentd-plugin-elasticsearch failing to connect to ES and crashing Fluentd worker

To Reproduce
Change ES_HOST DNS to an invalid address, so that requests to it will fail.

Expected behavior
FluentD worker should not be crashing when failing to connect to ElasticSearch and raising CPU usage to high levels.

Your Environment
FluentD v1.7.3 running in CentOS 7 Docker container on Linux (t3.medium EC2 instance)
Elasticsearch v6.3.1
Fluent-plugin-elasticsearch v3.5.2

Your Configuration

fluent.conf:
        <source>
          @type  forward
          @id    input1
          port  24224
        </source>

        <source>
          @type http_heartbeat
          port 8642
        </source>

        <filter docker.*.*>
          @type concat
          key log
          stream_identity_key container_id
          multiline_start_regexp /^\d{4}-\d{1,2}-\d{1,2}|^127.0.0.1|^{\"level\"/
          flush_mode interval
          flush_interval 10
          timeout_label @processdata
        </filter>

        <label @processdata>
          <match **>
            @type stdout
          </match>
        </label>

        <filter docker.*.*>
          @type parser
          key_name log
          <parse>
            @type json
            json_parser json
            time_key time
            time_type unixtime
            keep_time_key true
          </parse>
          replace_invalid_sequence true
          reserve_data true # this preserves unparsable log lines
          emit_invalid_record_to_error false # In case of unparsable log lines keep the error log clean
          reserve_time # the time was already parsed in the source, we don't want to overwrite it with current time.
        </filter>

        <filter docker.*.*>
          @type record_transformer
          enable_ruby true
          <record>
            formatted_time \${Time.at(record['time'].to_i/1000, record['time'].to_i%1000*1000).utc.strftime('%Y-%m-%dT%H:%M:%S.%LZ')}
            stack_name \"#{ENV['STACK_NAME']}\"
            log_level \${record['level'] == 10 ? 'trace' : record['level'] == 20 ? 'debug' : record['level'] == 30 ? 'info' : record['level'] == 40 ? 'warn' : record['level'] == 50 ? 'error' : 'fatal'}
          </record>
          remove_keys time,pid,v
        </filter>

        <match docker.*.*>
          @type copy
          <store ignore_error>
            @type elasticsearch
            @log_level debug
            host \"#{ENV['ES_HOST'] || 'es-logs'}\"
            port \"#{ENV['ES_PORT'] || 9200}\"
            logstash_prefix \"#{ENV['ES_LOGSTASH_PREFIX'] || 'logstash'}\"
            logstash_format true
            validate_client_version false
            type_name _doc
            request_timeout 15s
            sniffer_class_name 'Fluent::Plugin::ElasticsearchSimpleSniffer'
            reload_connections false
            reconnect_on_error true
            reload_on_failure true
            reload_after 100
            time_key formatted_time
            <buffer>
              @type file
              flush_mode interval
              flush_thread_count 4
              flush_interval 5s
              chunk_limit_size 8MB
              retry_wait 5
              retry_max_times 10
              retry_forever false
              retry_timeout 1h
              retry_type exponential_backoff
              overflow_action drop_oldest_chunk
              disable_chunk_backup true
              path /var/log/fluentd/buffer
            </buffer>
          </store>

          <store>
            @type file
            @id   output1
            path         /fluentd/log/\${tag[1]}/\${tag[2]}.%Y%m%d_%H
            append       true
            <buffer tag,time>
              @type memory
              flush_mode interval
              flush_interval 1s
              timekey 1h
              overflow_action drop_oldest_chunk
              disable_chunk_backup true
              retry_wait 5
              retry_max_times 10
              retry_forever false
              retry_timeout 1h
              retry_type exponential_backoff
            </buffer>
            <format>
              @type single_value
              message_key log
            </format>
            time_key formatted_time
          </store>
        </match>" >> /mnt/fluent.conf

Your Error Log

Repeated error below (in a constant loop):
2020-06-05 13:56:55 +0000 [info]: Worker 0 finished unexpectedly with status 1
2020-06-05 13:56:56 +0000 [info]: gem 'fluent-plugin-concat' version '2.4.0'
2020-06-05 13:56:56 +0000 [info]: gem 'fluent-plugin-elasticsearch' version '3.5.2'
2020-06-05 13:56:56 +0000 [info]: gem 'fluent-plugin-http-heartbeat' version '0.0.5'
2020-06-05 13:56:56 +0000 [info]: gem 'fluentd' version '1.7.3'
2020-06-05 13:56:56 +0000 [info]: adding match in @processdata pattern="**" type="stdout"
2020-06-05 13:56:56 +0000 [info]: adding filter pattern="docker.*.*" type="concat"
2020-06-05 13:56:56 +0000 [info]: adding filter pattern="docker.*.*" type="parser"
2020-06-05 13:56:56 +0000 [info]: adding filter pattern="docker.*.*" type="record_transformer"
2020-06-05 13:56:56 +0000 [info]: adding match pattern="docker.*.*" type="copy"
2020-06-05 13:56:56 +0000 [error]: #0 unexpected error error_class=NameError error="uninitialized constant Faraday::Error::ConnectionFailed\nDid you mean?  Faraday::ConnectionFailed"
  2020-06-05 13:56:56 +0000 [error]: #0 /ruby/2.3.1/lib/ruby/gems/2.3.0/gems/elasticsearch-transport-6.3.1/lib/elasticsearch/transport/transport/http/faraday.rb:46:in `host_unreachable_exceptions'
  2020-06-05 13:56:56 +0000 [error]: #0 /ruby/2.3.1/lib/ruby/gems/2.3.0/gems/fluent-plugin-elasticsearch-3.5.2/lib/fluent/plugin/elasticsearch_index_template.rb:36:in `rescue in retry_operate'
  2020-06-05 13:56:56 +0000 [error]: #0 /ruby/2.3.1/lib/ruby/gems/2.3.0/gems/fluent-plugin-elasticsearch-3.5.2/lib/fluent/plugin/elasticsearch_index_template.rb:34:in `retry_operate'
  2020-06-05 13:56:56 +0000 [error]: #0 /ruby/2.3.1/lib/ruby/gems/2.3.0/gems/fluent-plugin-elasticsearch-3.5.2/lib/fluent/plugin/out_elasticsearch.rb:244:in `configure'
  2020-06-05 13:56:56 +0000 [error]: #0 /ruby/2.3.1/lib/ruby/gems/2.3.0/gems/fluentd-1.7.3/lib/fluent/plugin.rb:164:in `configure'
  2020-06-05 13:56:56 +0000 [error]: #0 /ruby/2.3.1/lib/ruby/gems/2.3.0/gems/fluentd-1.7.3/lib/fluent/plugin/multi_output.rb:74:in `block in configure'
  2020-06-05 13:56:56 +0000 [error]: #0 /ruby/2.3.1/lib/ruby/gems/2.3.0/gems/fluentd-1.7.3/lib/fluent/plugin/multi_output.rb:63:in `each'
  2020-06-05 13:56:56 +0000 [error]: #0 /ruby/2.3.1/lib/ruby/gems/2.3.0/gems/fluentd-1.7.3/lib/fluent/plugin/multi_output.rb:63:in `configure'
  2020-06-05 13:56:56 +0000 [error]: #0 /ruby/2.3.1/lib/ruby/gems/2.3.0/gems/fluentd-1.7.3/lib/fluent/plugin/out_copy.rb:36:in `configure'
  2020-06-05 13:56:56 +0000 [error]: #0 /ruby/2.3.1/lib/ruby/gems/2.3.0/gems/fluentd-1.7.3/lib/fluent/plugin.rb:164:in `configure'
  2020-06-05 13:56:56 +0000 [error]: #0 /ruby/2.3.1/lib/ruby/gems/2.3.0/gems/fluentd-1.7.3/lib/fluent/agent.rb:130:in `add_match'
  2020-06-05 13:56:56 +0000 [error]: #0 /ruby/2.3.1/lib/ruby/gems/2.3.0/gems/fluentd-1.7.3/lib/fluent/agent.rb:72:in `block in configure'
  2020-06-05 13:56:56 +0000 [error]: #0 /ruby/2.3.1/lib/ruby/gems/2.3.0/gems/fluentd-1.7.3/lib/fluent/agent.rb:64:in `each'
  2020-06-05 13:56:56 +0000 [error]: #0 /ruby/2.3.1/lib/ruby/gems/2.3.0/gems/fluentd-1.7.3/lib/fluent/agent.rb:64:in `configure'
  2020-06-05 13:56:56 +0000 [error]: #0 /ruby/2.3.1/lib/ruby/gems/2.3.0/gems/fluentd-1.7.3/lib/fluent/root_agent.rb:150:in `configure'
  2020-06-05 13:56:56 +0000 [error]: #0 /ruby/2.3.1/lib/ruby/gems/2.3.0/gems/fluentd-1.7.3/lib/fluent/engine.rb:131:in `configure'
  2020-06-05 13:56:56 +0000 [error]: #0 /ruby/2.3.1/lib/ruby/gems/2.3.0/gems/fluentd-1.7.3/lib/fluent/engine.rb:96:in `run_configure'
  2020-06-05 13:56:56 +0000 [error]: #0 /ruby/2.3.1/lib/ruby/gems/2.3.0/gems/fluentd-1.7.3/lib/fluent/supervisor.rb:804:in `run_configure'
  2020-06-05 13:56:56 +0000 [error]: #0 /ruby/2.3.1/lib/ruby/gems/2.3.0/gems/fluentd-1.7.3/lib/fluent/supervisor.rb:550:in `block in run_worker'
  2020-06-05 13:56:56 +0000 [error]: #0 /ruby/2.3.1/lib/ruby/gems/2.3.0/gems/fluentd-1.7.3/lib/fluent/supervisor.rb:733:in `main_process'
  2020-06-05 13:56:56 +0000 [error]: #0 /ruby/2.3.1/lib/ruby/gems/2.3.0/gems/fluentd-1.7.3/lib/fluent/supervisor.rb:546:in `run_worker'
  2020-06-05 13:56:56 +0000 [error]: #0 /ruby/2.3.1/lib/ruby/gems/2.3.0/gems/fluentd-1.7.3/lib/fluent/command/fluentd.rb:320:in `<top (required)>'
  2020-06-05 13:56:56 +0000 [error]: #0 /ruby/2.3.1/lib/ruby/2.3.0/rubygems/core_ext/kernel_require.rb:68:in `require'
  2020-06-05 13:56:56 +0000 [error]: #0 /ruby/2.3.1/lib/ruby/2.3.0/rubygems/core_ext/kernel_require.rb:68:in `require'
  2020-06-05 13:56:56 +0000 [error]: #0 /ruby/2.3.1/lib/ruby/gems/2.3.0/gems/fluentd-1.7.3/bin/fluentd:8:in `<top (required)>'
  2020-06-05 13:56:56 +0000 [error]: #0 /ruby//2.3.1/bin/fluentd:23:in `load'
  2020-06-05 13:56:56 +0000 [error]: #0 /ruby//2.3.1/bin/fluentd:23:in `<main>'
2020-06-05 13:56:56 +0000 [info]: Worker 0 finished unexpectedly with status 1
2020-06-05 13:56:57 +0000 [info]: gem 'fluent-plugin-concat' version '2.4.0'
2020-06-05 13:56:57 +0000 [info]: gem 'fluent-plugin-elasticsearch' version '3.5.2'
2020-06-05 13:56:57 +0000 [info]: gem 'fluent-plugin-http-heartbeat' version '0.0.5'
2020-06-05 13:56:57 +0000 [info]: gem 'fluentd' version '1.7.3'
2020-06-05 13:56:57 +0000 [info]: adding match in @processdata pattern="**" type="stdout"
2020-06-05 13:56:57 +0000 [info]: adding filter pattern="docker.*.*" type="concat"
2020-06-05 13:56:57 +0000 [info]: adding filter pattern="docker.*.*" type="parser"
2020-06-05 13:56:57 +0000 [info]: adding filter pattern="docker.*.*" type="record_transformer"
2020-06-05 13:56:57 +0000 [info]: adding match pattern="docker.*.*" type="copy"
2020-06-05 13:56:57 +0000 [error]: #0 unexpected error error_class=NameError error="uninitialized constant Faraday::Error::ConnectionFailed\nDid you mean?  Faraday::ConnectionFailed"
  2020-06-05 13:56:57 +0000 [error]: #0 /ruby/2.3.1/lib/ruby/gems/2.3.0/gems/elasticsearch-transport-6.3.1/lib/elasticsearch/transport/transport/http/faraday.rb:46:in `host_unreachable_exceptions'
  2020-06-05 13:56:57 +0000 [error]: #0 /ruby/2.3.1/lib/ruby/gems/2.3.0/gems/fluent-plugin-elasticsearch-3.5.2/lib/fluent/plugin/elasticsearch_index_template.rb:36:in `rescue in retry_operate'
  2020-06-05 13:56:57 +0000 [error]: #0 /ruby/2.3.1/lib/ruby/gems/2.3.0/gems/fluent-plugin-elasticsearch-3.5.2/lib/fluent/plugin/elasticsearch_index_template.rb:34:in `retry_operate'
  2020-06-05 13:56:57 +0000 [error]: #0 /ruby/2.3.1/lib/ruby/gems/2.3.0/gems/fluent-plugin-elasticsearch-3.5.2/lib/fluent/plugin/out_elasticsearch.rb:244:in `configure'
  2020-06-05 13:56:57 +0000 [error]: #0 /ruby/2.3.1/lib/ruby/gems/2.3.0/gems/fluentd-1.7.3/lib/fluent/plugin.rb:164:in `configure'
  2020-06-05 13:56:57 +0000 [error]: #0 /ruby/2.3.1/lib/ruby/gems/2.3.0/gems/fluentd-1.7.3/lib/fluent/plugin/multi_output.rb:74:in `block in configure'
  2020-06-05 13:56:57 +0000 [error]: #0 /ruby/2.3.1/lib/ruby/gems/2.3.0/gems/fluentd-1.7.3/lib/fluent/plugin/multi_output.rb:63:in `each'
  2020-06-05 13:56:57 +0000 [error]: #0 /ruby/2.3.1/lib/ruby/gems/2.3.0/gems/fluentd-1.7.3/lib/fluent/plugin/multi_output.rb:63:in `configure'
  2020-06-05 13:56:57 +0000 [error]: #0 /ruby/2.3.1/lib/ruby/gems/2.3.0/gems/fluentd-1.7.3/lib/fluent/plugin/out_copy.rb:36:in `configure'
  2020-06-05 13:56:57 +0000 [error]: #0 /ruby/2.3.1/lib/ruby/gems/2.3.0/gems/fluentd-1.7.3/lib/fluent/plugin.rb:164:in `configure'
  2020-06-05 13:56:57 +0000 [error]: #0 /ruby/2.3.1/lib/ruby/gems/2.3.0/gems/fluentd-1.7.3/lib/fluent/agent.rb:130:in `add_match'
  2020-06-05 13:56:57 +0000 [error]: #0 /ruby/2.3.1/lib/ruby/gems/2.3.0/gems/fluentd-1.7.3/lib/fluent/agent.rb:72:in `block in configure'
  2020-06-05 13:56:57 +0000 [error]: #0 /ruby/2.3.1/lib/ruby/gems/2.3.0/gems/fluentd-1.7.3/lib/fluent/agent.rb:64:in `each'
  2020-06-05 13:56:57 +0000 [error]: #0 /ruby/2.3.1/lib/ruby/gems/2.3.0/gems/fluentd-1.7.3/lib/fluent/agent.rb:64:in `configure'
  2020-06-05 13:56:57 +0000 [error]: #0 /ruby/2.3.1/lib/ruby/gems/2.3.0/gems/fluentd-1.7.3/lib/fluent/root_agent.rb:150:in `configure'
  2020-06-05 13:56:57 +0000 [error]: #0 /ruby/2.3.1/lib/ruby/gems/2.3.0/gems/fluentd-1.7.3/lib/fluent/engine.rb:131:in `configure'

Additional context

[repeatedly]

When ES is down, FluentD CPU climbs to 50-70% usage, due to fluentd-plugin-elasticsearch failing to connect to ES and crashing Fluentd worker

So CPU spike doesn't happen on no error situation, right?

[williamzhao87]

When ES is down, FluentD CPU climbs to 50-70% usage, due to fluentd-plugin-elasticsearch failing to connect to ES and crashing Fluentd worker

So CPU spike doesn't happen on no error situation, right?

Correct, no CPU spikes in non-error situation

[cosmo0920]

elasticsearch-transport-6.3.1 should be used with faraday < 1 such that faraday v0.17.3.
This exception happens with faraday's breaking changes.
Could you confirm installed and using faraday gem version?

See: elastic/elasticsearch-ruby#752

[williamzhao87]

Confirmed that Faraday 1.0.0 is installed. How do we specify the dependency of elasticsearch-transport-6.3.1 to use faraday 0.17.3?

This is how it's currently installed in Dockerfile:
RUN gem install fluent-plugin-elasticsearch -v 3.5.2

[root@ip-10-64-8-168 gems]# gem list

*** LOCAL GEMS ***

bigdecimal (1.3.5, 1.2.8)
concurrent-ruby (1.1.5)
cool.io (1.5.4, 1.4.6)
did_you_mean (1.0.0)
dig_rb (1.0.1)
elasticsearch (6.3.1)
elasticsearch-api (6.3.1)
elasticsearch-transport (6.3.1)
excon (0.71.1)
faraday (1.0.0)
fluent-plugin-concat (2.4.0)
fluent-plugin-elasticsearch (3.5.2)
fluent-plugin-http-heartbeat (0.0.5)
fluentd (1.7.3)
http_parser.rb (0.6.0)
io-console (0.4.5)
json (2.2.0, 1.8.3)
minitest (5.8.3)
msgpack (1.3.1)
multi_json (1.14.1)
multipart-post (2.1.1)
net-telnet (0.1.1)
oj (3.3.10)
power_assert (0.2.6)
psych (2.0.17)
rake (10.4.2)
rdoc (4.2.1)
serverengine (2.2.0)
sigdump (0.2.4)
strptime (0.2.3)
test-unit (3.1.5)
tzinfo (2.0.1)
tzinfo-data (1.2019.3)
yajl-ruby (1.4.1)

[cosmo0920]

Confirmed that Faraday 1.0.0 is installed. How do we specify the dependency of elasticsearch-transport-6.3.1 to use faraday 0.17.3?

elasticsearch-transport 6.3.1 is a bit of old version.
The latest elasticsearch-transport 6.x series is elasticsearch-transport v6.8.2
This version should be using faraday 1.0.

Could you use version v6.8.2 of elasticsearch, elasticsearch-api, elasticsearch-transport?

[williamzhao87]

@cosmo0920
Thank you, upgrading ES version resolved the issue.