fluentd - Fluentd 重新部署后未将日志推送到 Elastic Search

Question

重新部署 Fluentd 集群后，日志暂时不会推送到 Elastic Search，有时需要数小时才能最终获取日志。

我的配置：

<source>
  @type systemd
  path /run/log/journal
  matches [{ "_SYSTEMD_UNIT": "docker.service" }]
  tag docker
  read_from_head true
</source>

<filter docker>
  @type record_transformer
  enable_ruby true
  <record>
    severity log
  </record>
</filter>

<match docker>
  @type elasticsearch
  host elasticsearch
  port ****
  scheme http
  request_timeout 30s
  http_backend_excon_nonblock false
  suppress_type_name true
  index_name test-%Y%m%d
  type_name docker-fluentd
   logstash_format true
  logstash_prefix test
  reconnect_on_error true
  reload_on_failure true
  reload_connections false
<buffer>
    @type file
    path /fluentd/log/elastic-buffer
    flush_thread_count 8
    flush_interval 1s
    chunk_limit_size 32M
    queue_limit_length 4
    flush_mode interval
    retry_max_interval 30
    retry_forever true
  </buffer>
</match>

我在日志中收到以下错误：

parsing config file is succeeded path="fluent.conf"
2021-03-23 13:23:51 +0000 [info]: gem 'fluent-plugin-concat' version '2.4.0'
2021-03-23 13:23:51 +0000 [info]: gem 'fluent-plugin-dedot_filter' version '1.0.0'
2021-03-23 13:23:51 +0000 [info]: gem 'fluent-plugin-detect-exceptions' version '0.0.13'
2021-03-23 13:23:51 +0000 [info]: gem 'fluent-plugin-elasticsearch' version '4.3.3'
2021-03-23 13:23:51 +0000 [info]: gem 'fluent-plugin-grok-parser' version '2.6.2'
2021-03-23 13:23:51 +0000 [info]: gem 'fluent-plugin-json-in-json-2' version '1.0.2'
2021-03-23 13:23:51 +0000 [info]: gem 'fluent-plugin-kubernetes_metadata_filter' version '2.6.0'
2021-03-23 13:23:51 +0000 [info]: gem 'fluent-plugin-multi-format-parser' version '1.0.0'
2021-03-23 13:23:51 +0000 [info]: gem 'fluent-plugin-prometheus' version '1.8.5'
2021-03-23 13:23:51 +0000 [info]: gem 'fluent-plugin-record-modifier' version '2.1.0'
2021-03-23 13:23:51 +0000 [info]: gem 'fluent-plugin-rewrite-tag-filter' version '2.4.0'
2021-03-23 13:23:51 +0000 [info]: gem 'fluent-plugin-systemd' version '1.0.2'
2021-03-23 13:23:51 +0000 [info]: gem 'fluentd' version '1.12.0'
2021-03-23 13:23:51 +0000 [warn]: both of Plugin @id and path for <storage> are not specified. Using on-memory store.
2021-03-23 13:23:51 +0000 [info]: using configuration file: <ROOT>


 
2021-03-23 13:23:51 +0000 [info]: starting fluentd-1.12.0 pid=8 ruby="2.6.6"
2021-03-23 13:23:51 +0000 [info]: spawn command to main:  cmdline=["/usr/local/bin/ruby", "-Eascii-8bit:ascii-8bit", "/fluentd/vendor/bundle/ruby/2.6.0/bin/fluentd", "-c", "/fluentd/etc/fluent.conf", "-p", "/fluentd/plugins", "--gemfile", "/fluentd/Gemfile", "-r", "/fluentd/vendor/bundle/ruby/2.6.0/gems/fluent-plugin-elasticsearch-4.3.3/lib/fluent/plugin/elasticsearch_simple_sniffer.rb", "--under-supervisor"]
2021-03-23 13:23:52 +0000 [info]: adding filter pattern="docker" type="record_transformer"
2021-03-23 13:23:52 +0000 [info]: adding match pattern="docker" type="elasticsearch"
2021-03-23 13:23:52 +0000 [info]: adding source type="systemd"
2021-03-23 13:23:52 +0000 [warn]: #0 both of Plugin @id and path for <storage> are not specified. Using on-memory store.
2021-03-23 13:23:52 +0000 [info]: #0 starting fluentd worker pid=16 ppid=8 worker=0
2021-03-23 13:23:52 +0000 [info]: #0 fluentd worker is now running worker=0
2021-03-23 13:26:57 +0000 [warn]: #0 failed to flush the buffer. retry_time=0 next_retry_seconds=2021-03-23 13:26:58 +0000 chunk="5be342287e58bb254e5c2b20d8deb496" error_class=Fluent::Plugin::ElasticsearchOutput::RecoverableRequestFailure error="could not push logs to Elasticsearch cluster ({:host=>\"elasticsearch\", :port=>9200, :scheme=>\"http\"}): Rejected Elasticsearch due to es_rejected_execution_exception"
    failed to flush the buffer. retry_time=0 next_retry_seconds=2021-03-23 13:06:00 +0000 chunk="5be33d724d11145f2605b016a9e6cd48" error_class=Fluent::Plugin::ElasticsearchOutput::RecoverableRequestFailure error="could not push logs to Elasticsearch cluster ({:host=>\"elasticsearch\", :port=>9200, :scheme=>\"http\"}): [429] {\"error\":{\"root_cause\":[{\"type\":\"es_rejected_execution_exception\",\"reason\":\"rejected execution of coordinating operation [coordinating_and_primary_bytes=44407079, replica_bytes=0, all_bytes=44407079, coordinating_operation_bytes=10894804, max_coordinating_and_primary_bytes=53687091]\"}],\"type\":\"es_rejected_execution_exception\",\"reason\":\"rejected execution of coordinating operation [coordinating_and_primary_bytes=44407079, replica_bytes=0, all_bytes=44407079, coordinating_operation_bytes=10894804, max_coordinating_and_primary_bytes=53687091]\"},\"status\":429}"

我在谷歌搜索时尝试了一些解决方案，但没有一个能解决我的问题。

弹性搜索界面：

弹性搜索

添加了更多日志

尝试修改过滤器以从现有字段中获取值时出现以下错误

流利的日志

标签： fluentd