Description
- logstash version: 7.17.1
- Java version: 17.0.1
In our case we read a billion of data per day from kafka, and we push this data to S3 throught logstash.
this is the pipeline config:
input {
kafka {
bootstrap_servers => "{{ msk_bootstrap_brokers }}"
codec => "json"
decorate_events => "extended"
group_id => "{{ msk_consumer_archive_consumer_group }}"
id => "{{ msk_consumer_archive_consumer_group }}"
partition_assignment_strategy => "cooperative_sticky"
security_protocol => SSL
ssl_truststore_location => "{{ msk_client_truststore_path }}"
# Use Filebeat topic and not filebeat_archive topic to avoid instance to duplicate message in topic
topics => ["{{ msk_topic_name_filebeat_odigo }}"]
}
}
output {
if [input][type] == "container" or [input][type] == "docker" {
s3 {
bucket => "{{ bucket_logging.name }}"
region => "{{ bucket_logging.region }}"
role_arn => "arn:aws:iam::{{ bucket_logging.account_id }}:role/{{ logstash_aws_role_name }}"
canned_acl => "bucket-owner-full-control"
validate_credentials_on_root_bucket => false
codec => "json_lines"
prefix => xxxxxxxxxxxxxxxxxxxxxxxxxxxx
temporary_directory => "/tmp/logstash/docker_uploads"
}
} else if [input][type] == "log" and [event][dataset] == "system.auth" {
s3 {
bucket => "{{ bucket_logging.name }}"
region => "{{ bucket_logging.region }}"
role_arn => "arn:aws:iam::{{ bucket_logging.account_id }}:role/{{ logstash_aws_role_name }}"
canned_acl => "bucket-owner-full-control"
validate_credentials_on_root_bucket => false
codec => "json_lines"
prefix => xxxxxxxxxxxxxxxxxxxx
temporary_directory => "/tmp/logstash/authlog_uploads"
}
} else {
s3 {
bucket => "{{ bucket_logging.name }}"
region => "{{ bucket_logging.region }}"
role_arn => "arn:aws:iam::{{ bucket_logging.account_id }}:role/{{ logstash_aws_role_name }}"
canned_acl => "bucket-owner-full-control"
validate_credentials_on_root_bucket => false
codec => "json_lines"
prefix => xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
temporary_directory => "/tmp/logstash/syslogs_uploads"
}
}
}
every morning we have this issue when there is one consumer and 2 millions of data to ingest:
[2023-09-20T00:02:21,944][ERROR][logstash.javapipeline ][consumer_archive_filebeat_odigo] Pipeline worker error, the pipeline will be stopped {:pipeline_id=>"consumer_archive_filebeat_odigo", :error=>"(EMFILE) Too many open files - Too many open files", :exception=>Java::OrgJrubyExceptions::SystemCallError, :backtrace=>["org.jruby.RubyIO.sysopen(org/jruby/RubyIO.java:1234)", "org.jruby.RubyFile.initialize(org/jruby/RubyFile.java:365)", "org.jruby.RubyIO.open(org/jruby/RubyIO.java:1156)", "usr.share.logstash.vendor.bundle.jruby.$2_dot_5_dot_0.gems.logstash_minus_output_minus_s3_minus_4_dot_3_dot_5.lib.logstash.outputs.s3.temporary_file_factory.new_file(/usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/logstash-output-s3-4.3.5/lib/logstash/outputs/s3/temporary_file_factory.rb:89)", "usr.share.logstash.vendor.bundle.jruby.$2_dot_5_dot_0.gems.logstash_minus_output_minus_s3_minus_4_dot_3_dot_5.lib.logstash.outputs.s3.temporary_file_factory.rotate!(/usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/logstash-output-s3-4.3.5/lib/logstash/outputs/s3/temporary_file_factory.rb:43)", "org.jruby.ext.thread.Mutex.synchronize(org/jruby/ext/thread/Mutex.java:164)", "usr.share.logstash.vendor.bundle.jruby.$2_dot_5_dot_0.gems.logstash_minus_output_minus_s3_minus_4_dot_3_dot_5.lib.logstash.outputs.s3.temporary_file_factory.rotate!(/usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/logstash-output-s3-4.3.5/lib/logstash/outputs/s3/temporary_file_factory.rb:42)", "usr.share.logstash.vendor.bundle.jruby.$2_dot_5_dot_0.gems.logstash_minus_output_minus_s3_minus_4_dot_3_dot_5.lib.logstash.outputs.s3.temporary_file_factory.initialize(/usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/logstash-output-s3-4.3.5/lib/logstash/outputs/s3/temporary_file_factory.rb:38)", "usr.share.logstash.vendor.bundle.jruby.$2_dot_5_dot_0.gems.logstash_minus_output_minus_s3_minus_4_dot_3_dot_5.lib.logstash.outputs.s3.file_repository.apply(/usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/logstash-output-s3-4.3.5/lib/logstash/outputs/s3/file_repository.rb:53)", "usr.share.logstash.vendor.bundle.jruby.$2_dot_5_dot_0.gems.logstash_minus_output_minus_s3_minus_4_dot_3_dot_5.lib.logstash.outputs.s3.file_repository.get_factory(/usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/logstash-output-s3-4.3.5/lib/logstash/outputs/s3/file_repository.rb:83)", "usr.share.logstash.vendor.bundle.jruby.$2_dot_5_dot_0.gems.logstash_minus_output_minus_s3_minus_4_dot_3_dot_5.lib.logstash.outputs.s3.file_repository.get_file(/usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/logstash-output-s3-4.3.5/lib/logstash/outputs/s3/file_repository.rb:87)", "usr.share.logstash.vendor.bundle.jruby.$2_dot_5_dot_0.gems.logstash_minus_output_minus_s3_minus_4_dot_3_dot_5.lib.logstash.outputs.s3.multi_receive_encoded(/usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/logstash-output-s3-4.3.5/lib/logstash/outputs/s3.rb:246)", "org.jruby.RubyArray.each(org/jruby/RubyArray.java:1821)", "usr.share.logstash.vendor.bundle.jruby.$2_dot_5_dot_0.gems.logstash_minus_output_minus_s3_minus_4_dot_3_dot_5.lib.logstash.outputs.s3.multi_receive_encoded(/usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/logstash-output-s3-4.3.5/lib/logstash/outputs/s3.rb:241)", "usr.share.logstash.logstash_minus_core.lib.logstash.outputs.base.multi_receive(/usr/share/logstash/logstash-core/lib/logstash/outputs/base.rb:103)", "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)", "usr.share.logstash.logstash_minus_core.lib.logstash.java_pipeline.start_workers(/usr/share/logstash/logstash-core/lib/logstash/java_pipeline.rb:299)"], :thread=>"#<Thread:0x6cde7172@/usr/share/logstash/logstash-core/lib/logstash/java_pipeline.rb:129 sleep>"} [2023-09-20T00:02:22,812][ERROR][logstash.javapipeline ][main] Pipeline worker error, the pipeline will be stopped {:pipeline_id=>"main", :error=>"(EMFILE) Too many open files - Too many open files", :exception=>Java::OrgJrubyExceptions::SystemCallError, :backtrace=>["org.jruby.RubyIO.sysopen(org/jruby/RubyIO.java:1234)", "org.jruby.RubyFile.initialize(org/jruby/RubyFile.java:365)", "org.jruby.RubyIO.open(org/jruby/RubyIO.java:1156)", "usr.share.logstash.vendor.bundle.jruby.$2_dot_5_dot_0.gems.logstash_minus_output_minus_s3_minus_4_dot_3_dot_5.lib.logstash.outputs.s3.temporary_file_factory.new_file(/usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/logstash-output-s3-4.3.5/lib/logstash/outputs/s3/temporary_file_factory.rb:89)", "usr.share.logstash.vendor.bundle.jruby.$2_dot_5_dot_0.gems.logstash_minus_output_minus_s3_minus_4_dot_3_dot_5.lib.logstash.outputs.s3.temporary_file_factory.rotate!(/usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/logstash-output-s3-4.3.5/lib/logstash/outputs/s3/temporary_file_factory.rb:43)", "org.jruby.ext.thread.Mutex.synchronize(org/jruby/ext/thread/Mutex.java:164)", "usr.share.logstash.vendor.bundle.jruby.$2_dot_5_dot_0.gems.logstash_minus_output_minus_s3_minus_4_dot_3_dot_5.lib.logstash.outputs.s3.temporary_file_factory.rotate!(/usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/logstash-output-s3-4.3.5/lib/logstash/outputs/s3/temporary_file_factory.rb:42)", "usr.share.logstash.vendor.bundle.jruby.$2_dot_5_dot_0.gems.logstash_minus_output_minus_s3_minus_4_dot_3_dot_5.lib.logstash.outputs.s3.temporary_file_factory.initialize(/usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/logstash-output-s3-4.3.5/lib/logstash/outputs/s3/temporary_file_factory.rb:38)", "usr.share.logstash.vendor.bundle.jruby.$2_dot_5_dot_0.gems.logstash_minus_output_minus_s3_minus_4_dot_3_dot_5.lib.logstash.outputs.s3.file_repository.apply(/usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/logstash-output-s3-4.3.5/lib/logstash/outputs/s3/file_repository.rb:53)", "usr.share.logstash.vendor.bundle.jruby.$2_dot_5_dot_0.gems.logstash_minus_output_minus_s3_minus_4_dot_3_dot_5.lib.logstash.outputs.s3.file_repository.get_factory(/usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/logstash-output-s3-4.3.5/lib/logstash/outputs/s3/file_repository.rb:83)", "usr.share.logstash.vendor.bundle.jruby.$2_dot_5_dot_0.gems.logstash_minus_output_minus_s3_minus_4_dot_3_dot_5.lib.logstash.outputs.s3.file_repository.get_file(/usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/logstash-output-s3-4.3.5/lib/logstash/outputs/s3/file_repository.rb:87)", "usr.share.logstash.vendor.bundle.jruby.$2_dot_5_dot_0.gems.logstash_minus_output_minus_s3_minus_4_dot_3_dot_5.lib.logstash.outputs.s3.multi_receive_encoded(/usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/logstash-output-s3-4.3.5/lib/logstash/outputs/s3.rb:246)", "org.jruby.RubyArray.each(org/jruby/RubyArray.java:1821)", "usr.share.logstash.vendor.bundle.jruby.$2_dot_5_dot_0.gems.logstash_minus_output_minus_s3_minus_4_dot_3_dot_5.lib.logstash.outputs.s3.multi_receive_encoded(/usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/logstash-output-s3-4.3.5/lib/logstash/outputs/s3.rb:241)", "usr.share.logstash.logstash_minus_core.lib.logstash.outputs.base.multi_receive(/usr/share/logstash/logstash-core/lib/logstash/outputs/base.rb:103)", "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)", "usr.share.logstash.logstash_minus_core.lib.logstash.java_pipeline.start_workers(/usr/share/logstash/logstash-core/lib/logstash/java_pipeline.rb:299)"], :thread=>"#<Thread:0x40bce2da@/usr/share/logstash/logstash-core/lib/logstash/java_pipeline.rb:129 sleep>"} [2023-09-20T00:02:22,826][ERROR][logstash.javapipeline ][main] Pipeline worker error, the pipeline will be stopped {:pipeline_id=>"main", :error=>"(EMFILE) Too many open files - Too many open files", :exception=>Java::OrgJrubyExceptions::SystemCallError, :backtrace=>["org.jruby.RubyIO.sysopen(org/jruby/RubyIO.java:1234)", "org.jruby.RubyFile.initialize(org/jruby/RubyFile.java:365)", "org.jruby.RubyIO.open(org/jruby/RubyIO.java:1156)", "usr.share.logstash.vendor.bundle.jruby.$2_dot_5_dot_0.gems.logstash_minus_output_minus_s3_minus_4_dot_3_dot_5.lib.logstash.outputs.s3.temporary_file_factory.new_file(/usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/logstash-output-s3-4.3.5/lib/logstash/outputs/s3/temporary_file_factory.rb:89)", "usr.share.logstash.vendor.bundle.jruby.$2_dot_5_dot_0.gems.logstash_minus_output_minus_s3_minus_4_dot_3_dot_5.lib.logstash.outputs.s3.temporary_file_factory.rotate!(/usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/logstash-output-s3-4.3.5/lib/logstash/outputs/s3/temporary_file_factory.rb:43)", "org.jruby.ext.thread.Mutex.synchronize(org/jruby/ext/thread/Mutex.java:164)", "usr.share.logstash.vendor.bundle.jruby.$2_dot_5_dot_0.gems.logstash_minus_output_minus_s3_minus_4_dot_3_dot_5.lib.logstash.outputs.s3.temporary_file_factory.rotate!(/usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/logstash-output-s3-4.3.5/lib/logstash/outputs/s3/temporary_file_factory.rb:42)", "usr.share.logstash.vendor.bundle.jruby.$2_dot_5_dot_0.gems.logstash_minus_output_minus_s3_minus_4_dot_3_dot_5.lib.logstash.outputs.s3.temporary_file_factory.initialize(/usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/logstash-output-s3-4.3.5/lib/logstash/outputs/s3/temporary_file_factory.rb:38)", "usr.share.logstash.vendor.bundle.jruby.$2_dot_5_dot_0.gems.logstash_minus_output_minus_s3_minus_4_dot_3_dot_5.lib.logstash.outputs.s3.file_repository.apply(/usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/logstash-output-s3-4.3.5/lib/logstash/outputs/s3/file_repository.rb:53)", "usr.share.logstash.vendor.bundle.jruby.$2_dot_5_dot_0.gems.logstash_minus_output_minus_s3_minus_4_dot_3_dot_5.lib.logstash.outputs.s3.file_repository.get_factory(/usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/logstash-output-s3-4.3.5/lib/logstash/outputs/s3/file_repository.rb:83)", "usr.share.logstash.vendor.bundle.jruby.$2_dot_5_dot_0.gems.logstash_minus_output_minus_s3_minus_4_dot_3_dot_5.lib.logstash.outputs.s3.file_repository.get_file(/usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/logstash-output-s3-4.3.5/lib/logstash/outputs/s3/file_repository.rb:87)", "usr.share.logstash.vendor.bundle.jruby.$2_dot_5_dot_0.gems.logstash_minus_output_minus_s3_minus_4_dot_3_dot_5.lib.logstash.outputs.s3.multi_receive_encoded(/usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/logstash-output-s3-4.3.5/lib/logstash/outputs/s3.rb:246)", "org.jruby.RubyArray.each(org/jruby/RubyArray.java:1821)", "usr.share.logstash.vendor.bundle.jruby.$2_dot_5_dot_0.gems.logstash_minus_output_minus_s3_minus_4_dot_3_dot_5.lib.logstash.outputs.s3.multi_receive_encoded(/usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/logstash-output-s3-4.3.5/lib/logstash/outputs/s3.rb:241)", "usr.share.logstash.logstash_minus_core.lib.logstash.outputs.base.multi_receive(/usr/share/logstash/logstash-core/lib/logstash/outputs/base.rb:103)", "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)", "usr.share.logstash.logstash_minus_core.lib.logstash.java_pipeline.start_workers(/usr/share/logstash/logstash-core/lib/logstash/java_pipeline.rb:299)"], :thread=>"#<Thread:0x40bce2da@/usr/share/logstash/logstash-core/lib/logstash/java_pipeline.rb:129 sleep>"}
I think there is a lot of temporary file created in the temporary directory.
thanks