Logstash to fluentd

[rhysxevans]

Hi

Does anyone know the best way to send logs from logstash to fluentd.

I have tried tcp, but performance is slow.

Also https://github.com/tagomoris/logstash-output-fluentd seems to work and gives better performance, but crashes.

Any help is appreciated

Thanks

[rhysxevans]

There is a reason for this setup, due to beats issues, and at present this connection happens via 127.0.0.1

[rhysxevans]

Error when using logstash-output-fluentd

[2023-01-23T22:02:37,333][ERROR][logstash.javapipeline    ][Beats] Pipeline worker error, the pipeline will be stopped {:pipeline_id=>"Beats", :error=>"(NoMethodError) undefined method `to_msgpack' for 0.1e-2:BigDecimal", :exception=>Java::OrgJrubyExceptions::NoMethodError, :backtrace=>["org.msgpack.jruby.Packer.write(org/msgpack/jruby/Packer.java:128)", "opt.logstash_minus_7_dot_17_dot_5.vendor.bundle.jruby.$2_dot_5_dot_0.gems.msgpack_minus_1_dot_4_dot_5_minus_java.lib.msgpack.pack(/opt/logstash-7.17.5/vendor/bundle/jruby/2.5.0/gems/msgpack-1.4.5-java/lib/msgpack.rb:41)", "opt.logstash_minus_7_dot_17_dot_5.vendor.bundle.jruby.$2_dot_5_dot_0.gems.msgpack_minus_1_dot_4_dot_5_minus_java.lib.msgpack.core_ext.to_msgpack(/opt/logstash-7.17.5/vendor/bundle/jruby/2.5.0/gems/msgpack-1.4.5-java/lib/msgpack/core_ext.rb:11)", "opt.logstash_minus_7_dot_17_dot_5.vendor.bundle.jruby.$2_dot_5_dot_0.gems.logstash_minus_output_minus_fluentd_minus_1_dot_0_dot_0.lib.logstash.outputs.fluentd.convert_from_event_to_msgpack(/opt/logstash-7.17.5/vendor/bundle/jruby/2.5.0/gems/logstash-output-fluentd-1.0.0/lib/logstash/outputs/fluentd.rb:41)", "opt.logstash_minus_7_dot_17_dot_5.vendor.bundle.jruby.$2_dot_5_dot_0.gems.logstash_minus_output_minus_fluentd_minus_1_dot_0_dot_0.lib.logstash.outputs.fluentd.receive(/opt/logstash-7.17.5/vendor/bundle/jruby/2.5.0/gems/logstash-output-fluentd-1.0.0/lib/logstash/outputs/fluentd.rb:58)", "opt.logstash_minus_7_dot_17_dot_5.logstash_minus_core.lib.logstash.outputs.base.multi_receive(/opt/logstash-7.17.5/logstash-core/lib/logstash/outputs/base.rb:105)", "org.jruby.RubyArray.each(org/jruby/RubyArray.java:1821)", "opt.logstash_minus_7_dot_17_dot_5.logstash_minus_core.lib.logstash.outputs.base.multi_receive(/opt/logstash-7.17.5/logstash-core/lib/logstash/outputs/base.rb:105)", "org.logstash.config.ir.compiler.OutputStrategyExt$AbstractOutputStrategyExt.multi_receive(org/logstash/config/ir/compiler/OutputStrategyExt.java:143)", "org.logstash.config.ir.compiler.AbstractOutputDelegatorExt.multi_receive(org/logstash/config/ir/compiler/AbstractOutputDelegatorExt.java:121)", "opt.logstash_minus_7_dot_17_dot_5.logstash_minus_core.lib.logstash.java_pipeline.start_workers(/opt/logstash-7.17.5/logstash-core/lib/logstash/java_pipeline.rb:300)"], :thread=>"#<Thread:0x399f1fa6@/opt/logstash-7.17.5/logstash-core/lib/logstash/pipeline_action/create.rb:54 run>"}

[rhysxevans]

Sanitised config

Fluentd input

<source>
  @type forward
  port 5170
  bind 127.0.0.1
  tag logstash
</source>

Logstash Output
logstash.yml

monitoring.enabled: false
monitoring.cluster_uuid: ssssssss

pipeline.batch.size: 1000
pipeline.batch.delay: 50
pipeline.workers: "3"

path.data: /logstash
http.host: "0.0.0.0"
log.level: info
path.logs: /var/log/logstash

pipelines.yml

- pipeline.id: Beats
  queue.type: persisted
  queue.max_bytes: "512mb"
  path.queue: "/logstash"
  path.config: "/opt/logstash/config/beats.conf"
  pipeline.ecs_compatibility: v1

beats.conf

input {
  beats {
    port => 5044
  }
}
filter {
  ruby {
    code => "event.set('[arrival_timestamp]', Time.now());"
  }
}
output {
  fluentd {
    host => "127.0.0.1"
    port => 5170
    flush_size => 1000
  }
}