FluentD将日志从kafka转发到另一个fluentD

Question

我需要将我的应用程序日志发送到FluentD，它是EFK服务的一部分。因此，我尝试配置另一个FluentD来实现这一点。

my-fluent.conf:

<source>
  @type kafka_group
  consumer_group cgrp
  brokers "#{ENV['KAFKA_BROKERS']}"
  scram_mechanism sha512
  username "#{ENV['KAFKA_USERNAME']}"
  password "#{ENV['KAFKA_PASSWORD']}"
  ssl_ca_certs_from_system true
  topics "#{ENV['KAFKA_TOPICS']}"
  format json
</source>
<filter TOPIC>
  @type parser
  key_name log 
  reserve_data false
  <parse>
    @type json
  </parse>
</filter>
<match TOPIC>
  @type copy
  <store>
    @type stdout
  </store>
  <store>
    @type forward
    <server>
      host "#{ENV['FLUENTD_HOST']}"
      port "#{ENV['FLUENTD_PORT']}"
      shared_key "#{ENV['FLUENTD_SHARED_KEY']}"
    </server>
  </store>
</match>

我能够正确地看到stdout的输出

2021-07-06 07:36:54.376459650 +0000主题：{"foo":"bar"，.}

但我看不见基巴纳的原木。跟踪之后，我发现第二个流畅性是在接收数据时抛出错误：

{“时间”：“2021-07-05 11:21:41 +0000”、“级别”：“错误”、“消息”：“读取数据时的意外错误”X.X“port=58548 error_class=MessagePack::MalformedFormatError error=”无效字节“、"worker_id":0} {”时间“：”2021-07-05 11:21:41 +0000“、”级别“：”错误“、"worker_id":0，"message":"/usr/lib/ruby/gems/2.7.0/gems/fluentd-1.12.2/lib/fluent/plugin/in_forward.rb:262:in _read_ feed_each'\n/usr/lib/ruby/gems/2.7.0/gems/fluentd-1.12.2/lib/fluent/plugin/in_forward.rb:262:in _feed_each'\n/usr/lib/ruby/gems/2.7.0/gems/fluentd-1.12.2/lib/fluent/plugin/in_forward.rb:262:in‘\n/usr/lib/ruby上的read_messages'\n/usr/lib/ruby/gems/2.7.0/gems/fluentd-1.12.2/lib/fluent/plugin/in_forward.rb:271:in block in read_messages'\n/usr/lib/ruby/gems/2.7.0/gems/fluentd-1.12.2/lib/fluent/plugin_helper/server.rb:613:in中的feed_each'\n/usr/lib/ruby/gems/2.7.0/gems/fluentd-1.12.2/lib/fluent/plugin/in_forward.rb:262:in块(2个级别)/gems/2.7.0/gems/cool.io-1.7.1/lib/cool.io/io.rb:123:in on_readable'\n/usr/lib/ruby/gems/2.7.0/gems/cool.io-1.7.1/lib/cool.io/io.rb:186:in on_readable'\n/usr/lib/ruby/gems/2.7.0/gems/cool.io-1.7.1/lib/cool.io/loop.rb:88:in run_once'\n/usr/lib/ruby/gems/2.7.0/gems/cool.io-1.7.1/lib/cool.io/loop.rb:88:in run'\n/usr/lib/ruby/gems/2.7.0/gems/fluentd-1.12.2/lib/fluent/plugin_helper/Event循环系统.93:93: in block in start'\n/usr/lib/ruby/gems/2.7.0/gems/fluentd-1.12.2/lib/fluent/plugin_helper/thread.rb:78:in块中的thread_create‘’“}

Answer 1

问题是第一次漏掉了安全标签。

<match TOPIC>
  @type copy
  <store>
    @type stdout
  </store>
  <store>
    @type forward
    <server>
      host "#{ENV['FLUENTD_HOST']}"
      port "#{ENV['FLUENTD_PORT']}"
      shared_key "#{ENV['FLUENTD_SHARED_KEY']}"
    </server>
    <security>
      self_hostname HOSTNAME
      shared_key "#{ENV['FLUENTD_SHARED_KEY']}"
    </security>
  </store>
</match>