I want to use multiple pipelines for logstash for example for input section i want to have pipeline.id: source_input and for output and filter section i want to have pipeline.id: destination_output
But i am facing this error
[2021-03-18T11:44:29,804][INFO ][logstash.javapipeline ][packetbeat] Starting pipeline {:pipeline_id=>"packetbeat", "pipeline.workers"=>4, "pipeline.batch.size"=>125, "pipeline.batch.delay"=>50, "pipeline.max_inflight"=>500, "pipeline.sources"=>["/usr/share/logstash/pipeline/test/test.yaml"], :thread=>"#<Thread:0x70a53900 run>"}
[2021-03-18T11:44:29,818][ERROR][logstash.javapipeline ][packetbeat] Worker loop initialization error {:pipeline_id=>"packetbeat", :error=>"Missing Filter End Vertex", :exception=>Java::JavaLang::IllegalStateException, :stacktrace=>"org.logstash.config.ir.CompiledPipeline$CompiledExecution.lambda$compileFilters$1(org/logstash/config/ir/CompiledPipeline.java:376)\njava.util.Optional.orElseThrow(java/util/Optional.java:408)\norg.logstash.config.ir.CompiledPipeline$CompiledExecution.compileFilters(org/logstash/config/ir/CompiledPipeline.java:376)\norg.logstash.config.ir.CompiledPipeline$CompiledExecution.<init>(org/logstash/config/ir/CompiledPipeline.java:359)\norg.logstash.config.ir.CompiledPipeline$CompiledUnorderedExecution.<init>(org/logstash/config/ir/CompiledPipeline.java:319)\norg.logstash.config.ir.CompiledPipeline.buildExecution(org/logstash/config/ir/CompiledPipeline.java:153)\norg.logstash.execution.WorkerLoop.<init>(org/logstash/execution/WorkerLoop.java:63)\njdk.internal.reflect.GeneratedConstructorAccessor60.newInstance(jdk/internal/reflect/GeneratedConstructorAccessor60)\njdk.internal.reflect.DelegatingConstructorAccessorImpl.newInstance(jdk/internal/reflect/DelegatingConstructorAccessorImpl.java:45)\njava.lang.reflect.Constructor.newInstance(java/lang/reflect/Constructor.java:490)\norg.jruby.javasupport.JavaConstructor.newInstanceDirect(org/jruby/javasupport/JavaConstructor.java:253)\norg.jruby.RubyClass.newInstance(org/jruby/RubyClass.java:939)\norg.jruby.RubyClass$INVOKER$i$newInstance.call(org/jruby/RubyClass$INVOKER$i$newInstance.gen)\nusr.share.logstash.logstash_minus_core.lib.logstash.java_pipeline.init_worker_loop(/usr/share/logstash/logstash-core/lib/logstash/java_pipeline.rb:565)\nusr.share.logstash.logstash_minus_core.lib.logstash.java_pipeline.RUBY$method$init_worker_loop$0$__VARARGS__(usr/share/logstash/logstash_minus_core/lib/logstash//usr/share/logstash/logstash-core/lib/logstash/java_pipeline.rb)\nusr.share.logstash.logstash_minus_core.lib.logstash.java_pipeline.start_workers(/usr/share/logstash/logstash-core/lib/logstash/java_pipeline.rb:280)\norg.jruby.RubyProc.call(org/jruby/RubyProc.java:318)\njava.lang.Thread.run(java/lang/Thread.java:834)", :thread=>"#<Thread:0x70a53900 run>"}
[2021-03-18T11:44:29,821][ERROR][logstash.javapipeline ][packetbeat] Worker loop initialization error {:pipeline_id=>"packetbeat", :error=>"Missing Filter End Vertex", :exception=>Java::JavaLang::IllegalStateException, :stacktrace=>"org.logstash.config.ir.CompiledPipeline$CompiledExecution.lambda$compileFilters$1(org/logstash/config/ir/CompiledPipeline.java:376)\njava.util.Optional.orElseThrow(java/util/Optional.java:408)\norg.logstash.config.ir.CompiledPipeline$CompiledExecution.compileFilters(org/logstash/config/ir/CompiledPipeline.java:376)\norg.logstash.config.ir.CompiledPipeline$CompiledExecution.<init>(org/logstash/config/ir/CompiledPipeline.java:359)\norg.logstash.config.ir.CompiledPipeline$CompiledUnorderedExecution.<init>(org/logstash/config/ir/CompiledPipeline.java:319)\norg.logstash.config.ir.CompiledPipeline.buildExecution(org/logstash/config/ir/CompiledPipeline.java:153)\norg.logstash.execution.WorkerLoop.<init>(org/logstash/execution/WorkerLoop.java:63)\njdk.internal.reflect.GeneratedConstructorAccessor60.newInstance(jdk/internal/reflect/GeneratedConstructorAccessor60)\njdk.internal.reflect.DelegatingConstructorAccessorImpl.newInstance(jdk/internal/reflect/DelegatingConstructorAccessorImpl.java:45)\njava.lang.reflect.Constructor.newInstance(java/lang/reflect/Constructor.java:490)\norg.jruby.javasupport.JavaConstructor.newInstanceDirect(org/jruby/javasupport/JavaConstructor.java:253)\norg.jruby.RubyClass.newInstance(org/jruby/RubyClass.java:939)\norg.jruby.RubyClass$INVOKER$i$newInstance.call(org/jruby/RubyClass$INVOKER$i$newInstance.gen)\nusr.share.logstash.logstash_minus_core.lib.logstash.java_pipeline.init_worker_loop(/usr/share/logstash/logstash-core/lib/logstash/java_pipeline.rb:565)\nusr.share.logstash.logstash_minus_core.lib.logstash.java_pipeline.RUBY$method$init_worker_loop$0$__VARARGS__(usr/share/logstash/logstash_minus_core/lib/logstash//usr/share/logstash/logstash-core/lib/logstash/java_pipeline.rb)\nusr.share.logstash.logstash_minus_core.lib.logstash.java_pipeline.start_workers(/usr/share/logstash/logstash-core/lib/logstash/java_pipeline.rb:280)\norg.jruby.RubyProc.call(org/jruby/RubyProc.java:318)\njava.lang.Thread.run(java/lang/Thread.java:834)", :thread=>"#<Thread:0x70a53900 run>"}
pipeline.yaml
- pipeline.id: beats_input
path.config: "/usr/share/logstash/pipeline/beats_input/"
pipeline.batch.size: 2000
- pipeline.id: packetbeat
path.config: "/usr/share/logstash/pipeline/packetbeat/"
queue.type: persisted
In packetbeat directory i have filter+output files and in beats_input directory i have input file
if i move input file to packetbeat and remove pipeline.id: beats_input section then it will start working