Logstash remains in running state

Hi ,

I want to setup a logstash pipeline which will read data from Elasticsearch and load it to s3. For time being i am creating a test pipeline which reads data from one file and create a output file from it (creating dulplicate file) but thing is when i trigger my run it keeps in the running state and didn't execute as expected

my configuration file looks like :

input { 
	file {
	 path => "/Users/ritik.loomba/Downloads/sample.json" }
} 
output { 
	file { 
 	 path => "/Users/ritik.loomba/Downloads/unload.json"}
}

and messages i am getting is :

bin/logstash -f config/filetofile.conf
Using bundled JDK: /Users/ritik.loomba/Documents/ELKStack/logstash-8.16.1/jdk.app/Contents/Home
Sending Logstash logs to /Users/ritik.loomba/Documents/ELKStack/logstash-8.16.1/logs which is now configured via log4j2.properties
[2024-12-02T17:55:54,549][INFO ][logstash.runner          ] Log4j configuration path used is: /Users/ritik.loomba/Documents/ELKStack/logstash-8.16.1/config/log4j2.properties
[2024-12-02T17:55:54,551][INFO ][logstash.runner          ] Starting Logstash {"logstash.version"=>"8.16.1", "jruby.version"=>"jruby 9.4.9.0 (3.1.4) 2024-11-04 547c6b150e OpenJDK 64-Bit Server VM 21.0.5+11-LTS on 21.0.5+11-LTS +indy +jit [arm64-darwin]"}
[2024-12-02T17:55:54,552][INFO ][logstash.runner          ] JVM bootstrap flags: [-Xms1g, -Xmx1g, -Djava.awt.headless=true, -Dfile.encoding=UTF-8, -Djruby.compile.invokedynamic=true, -XX:+HeapDumpOnOutOfMemoryError, -Djava.security.egd=file:/dev/urandom, -Dlog4j2.isThreadContextMapInheritable=true, -Dlogstash.jackson.stream-read-constraints.max-string-length=200000000, -Dlogstash.jackson.stream-read-constraints.max-number-length=10000, -Djruby.regexp.interruptible=true, -Djdk.io.File.enableADS=true, --add-exports=jdk.compiler/com.sun.tools.javac.api=ALL-UNNAMED, --add-exports=jdk.compiler/com.sun.tools.javac.file=ALL-UNNAMED, --add-exports=jdk.compiler/com.sun.tools.javac.parser=ALL-UNNAMED, --add-exports=jdk.compiler/com.sun.tools.javac.tree=ALL-UNNAMED, --add-exports=jdk.compiler/com.sun.tools.javac.util=ALL-UNNAMED, --add-opens=java.base/java.security=ALL-UNNAMED, --add-opens=java.base/java.io=ALL-UNNAMED, --add-opens=java.base/java.nio.channels=ALL-UNNAMED, --add-opens=java.base/sun.nio.ch=ALL-UNNAMED, --add-opens=java.management/sun.management=ALL-UNNAMED, -Dio.netty.allocator.maxOrder=11]
[2024-12-02T17:55:54,553][INFO ][logstash.runner          ] Jackson default value override `logstash.jackson.stream-read-constraints.max-string-length` configured to `200000000`
[2024-12-02T17:55:54,553][INFO ][logstash.runner          ] Jackson default value override `logstash.jackson.stream-read-constraints.max-number-length` configured to `10000`
[2024-12-02T17:55:54,564][WARN ][logstash.config.source.multilocal] Ignoring the 'pipelines.yml' file because modules or command line options are specified
[2024-12-02T17:55:54,725][INFO ][logstash.agent           ] Successfully started Logstash API endpoint {:port=>9600, :ssl_enabled=>false}
[2024-12-02T17:55:54,793][INFO ][org.reflections.Reflections] Reflections took 41 ms to scan 1 urls, producing 149 keys and 523 values
[2024-12-02T17:55:54,863][INFO ][logstash.codecs.jsonlines] ECS compatibility is enabled but `target` option was not specified. This may cause fields to be set at the top-level of the event where they are likely to clash with the Elastic Common Schema. It is recommended to set the `target` option to avoid potential schema conflicts (if your data is ECS compliant or non-conflicting, feel free to ignore this message)
[2024-12-02T17:55:54,869][INFO ][logstash.javapipeline    ] Pipeline `main` is configured with `pipeline.ecs_compatibility: v8` setting. All plugins in this pipeline will default to `ecs_compatibility => v8` unless explicitly configured otherwise.
[2024-12-02T17:55:54,878][INFO ][logstash.javapipeline    ][main] Starting pipeline {:pipeline_id=>"main", "pipeline.workers"=>8, "pipeline.batch.size"=>125, "pipeline.batch.delay"=>50, "pipeline.max_inflight"=>1000, "pipeline.sources"=>["/Users/ritik.loomba/Documents/ELKStack/logstash-8.16.1/config/filetofile.conf"], :thread=>"#<Thread:0xbba223b /Users/ritik.loomba/Documents/ELKStack/logstash-8.16.1/logstash-core/lib/logstash/java_pipeline.rb:139 run>"}
[2024-12-02T17:55:55,062][INFO ][logstash.javapipeline    ][main] Pipeline Java execution initialization time {"seconds"=>0.18}
[2024-12-02T17:55:55,066][INFO ][logstash.inputs.file     ][main] No sincedb_path set, generating one based on the "path" setting {:sincedb_path=>"/Users/ritik.loomba/Documents/ELKStack/logstash-8.16.1/data/plugins/inputs/file/.sincedb_0e5b6dc105fb85a0435282afbff2a7b2", :path=>["/Users/ritik.loomba/Downloads/sample.json"]}
[2024-12-02T17:55:55,067][INFO ][logstash.javapipeline    ][main] Pipeline started {"pipeline.id"=>"main"}
[2024-12-02T17:55:55,068][INFO ][filewatch.observingtail  ][main][7b7c651e38eb3178a9477999377170c7f6fcf530140ac74c72974b5ccd08f6f6] START, creating Discoverer, Watch with file and sincedb collections
[2024-12-02T17:55:55,073][INFO ][logstash.agent           ] Pipelines running {:count=>1, :running_pipelines=>[:main], :non_running_pipelines=>[]}

I am using macOS, logstash-8.16.1

By default, Logstash will not exit after reading your file. Instead it will monitor the file for additional lines to forward to Elasticsearch.

To exit after reading the file add exit_after_read => true to your input.

Just to add that the file input needs to be using the read mode and even that, it may not always work.

There is an issue about it here

1 Like

I have tried both parameters:

input { 
	file {
	 path => "/Users/ritik.loomba/Downloads/sample.json" 
     exit_after_read => true
	 mode => "read"
}
} 
output { 
	file { 
 	 path => "/Users/ritik.loomba/Downloads/unload.json"}
}

but still getting same issue

Are you expecting it to exit after reading the file or is something else not working?

Yes, it would exit and create a file unload.json

My main requirement is to create a logstash pipeline which will unload data from elasticsearch to aws s3 in json format and then i will move that data to snowflake. Will do elasticsearch to s3 using logstash and rest will be handle in snowflake connector.

  1. Can we capture records increamentally ?
  2. does that capture updates and deletes also, if yes how ?

Currently i am trying in local environment,

input {
      elasticsearch {
        hosts => [ "https://localhost:9200" ]
        user => "user"
        password => "password"
        query => '{ "query": { "match": { "statuscode": 200 } }, "sort": [ "_doc" ] }'
      }
    }
output { 
	file { 
 	 path => "/Users/ritik.loomba/Downloads/unload.json"}
} 

and getting this message:

Using bundled JDK: /Users/ritik.loomba/Documents/ELKStack/logstash-8.16.1/jdk.app/Contents/Home
Sending Logstash logs to /Users/ritik.loomba/Documents/ELKStack/logstash-8.16.1/logs which is now configured via log4j2.properties
[2024-12-03T12:21:33,857][INFO ][logstash.runner          ] Log4j configuration path used is: /Users/ritik.loomba/Documents/ELKStack/logstash-8.16.1/config/log4j2.properties
[2024-12-03T12:21:33,860][INFO ][logstash.runner          ] Starting Logstash {"logstash.version"=>"8.16.1", "jruby.version"=>"jruby 9.4.9.0 (3.1.4) 2024-11-04 547c6b150e OpenJDK 64-Bit Server VM 21.0.5+11-LTS on 21.0.5+11-LTS +indy +jit [arm64-darwin]"}
[2024-12-03T12:21:33,861][INFO ][logstash.runner          ] JVM bootstrap flags: [-Xms1g, -Xmx1g, -Djava.awt.headless=true, -Dfile.encoding=UTF-8, -Djruby.compile.invokedynamic=true, -XX:+HeapDumpOnOutOfMemoryError, -Djava.security.egd=file:/dev/urandom, -Dlog4j2.isThreadContextMapInheritable=true, -Dlogstash.jackson.stream-read-constraints.max-string-length=200000000, -Dlogstash.jackson.stream-read-constraints.max-number-length=10000, -Djruby.regexp.interruptible=true, -Djdk.io.File.enableADS=true, --add-exports=jdk.compiler/com.sun.tools.javac.api=ALL-UNNAMED, --add-exports=jdk.compiler/com.sun.tools.javac.file=ALL-UNNAMED, --add-exports=jdk.compiler/com.sun.tools.javac.parser=ALL-UNNAMED, --add-exports=jdk.compiler/com.sun.tools.javac.tree=ALL-UNNAMED, --add-exports=jdk.compiler/com.sun.tools.javac.util=ALL-UNNAMED, --add-opens=java.base/java.security=ALL-UNNAMED, --add-opens=java.base/java.io=ALL-UNNAMED, --add-opens=java.base/java.nio.channels=ALL-UNNAMED, --add-opens=java.base/sun.nio.ch=ALL-UNNAMED, --add-opens=java.management/sun.management=ALL-UNNAMED, -Dio.netty.allocator.maxOrder=11]
[2024-12-03T12:21:33,862][INFO ][logstash.runner          ] Jackson default value override `logstash.jackson.stream-read-constraints.max-string-length` configured to `200000000`
[2024-12-03T12:21:33,862][INFO ][logstash.runner          ] Jackson default value override `logstash.jackson.stream-read-constraints.max-number-length` configured to `10000`
[2024-12-03T12:21:33,877][WARN ][logstash.config.source.multilocal] Ignoring the 'pipelines.yml' file because modules or command line options are specified
[2024-12-03T12:21:34,057][INFO ][logstash.agent           ] Successfully started Logstash API endpoint {:port=>9600, :ssl_enabled=>false}
[2024-12-03T12:21:34,138][INFO ][org.reflections.Reflections] Reflections took 45 ms to scan 1 urls, producing 149 keys and 523 values
[2024-12-03T12:21:34,552][INFO ][logstash.codecs.jsonlines] ECS compatibility is enabled but `target` option was not specified. This may cause fields to be set at the top-level of the event where they are likely to clash with the Elastic Common Schema. It is recommended to set the `target` option to avoid potential schema conflicts (if your data is ECS compliant or non-conflicting, feel free to ignore this message)
[2024-12-03T12:21:34,557][INFO ][logstash.javapipeline    ] Pipeline `main` is configured with `pipeline.ecs_compatibility: v8` setting. All plugins in this pipeline will default to `ecs_compatibility => v8` unless explicitly configured otherwise.
[2024-12-03T12:21:34,564][INFO ][logstash.javapipeline    ][main] Starting pipeline {:pipeline_id=>"main", "pipeline.workers"=>8, "pipeline.batch.size"=>125, "pipeline.batch.delay"=>50, "pipeline.max_inflight"=>1000, "pipeline.sources"=>["/Users/ritik.loomba/Documents/ELKStack/logstash-8.16.1/config/elasticsearchtofile.conf"], :thread=>"#<Thread:0x64eaad30 /Users/ritik.loomba/Documents/ELKStack/logstash-8.16.1/logstash-core/lib/logstash/java_pipeline.rb:139 run>"}
[2024-12-03T12:21:34,777][INFO ][logstash.javapipeline    ][main] Pipeline Java execution initialization time {"seconds"=>0.21}
[2024-12-03T12:21:36,568][ERROR][logstash.javapipeline    ][main] Pipeline error {:pipeline_id=>"main", :exception=>#<Manticore::ClientProtocolException: PKIX path building failed: sun.security.provider.certpath.SunCertPathBuilderException: unable to find valid certification path to requested target>, :backtrace=>["/Users/ritik.loomba/Documents/ELKStack/logstash-8.16.1/vendor/bundle/jruby/3.1.0/gems/manticore-0.9.1-java/lib/manticore/response.rb:36:in `block in initialize'", "/Users/ritik.loomba/Documents/ELKStack/logstash-8.16.1/vendor/bundle/jruby/3.1.0/gems/manticore-0.9.1-java/lib/manticore/response.rb:79:in `call'", "/Users/ritik.loomba/Documents/ELKStack/logstash-8.16.1/vendor/bundle/jruby/3.1.0/gems/manticore-0.9.1-java/lib/manticore/response.rb:274:in `call_once'", "/Users/ritik.loomba/Documents/ELKStack/logstash-8.16.1/vendor/bundle/jruby/3.1.0/gems/manticore-0.9.1-java/lib/manticore/response.rb:158:in `code'", "/Users/ritik.loomba/Documents/ELKStack/logstash-8.16.1/vendor/bundle/jruby/3.1.0/gems/elasticsearch-transport-7.17.11/lib/elasticsearch/transport/transport/http/manticore.rb:111:in `block in perform_request'", "/Users/ritik.loomba/Documents/ELKStack/logstash-8.16.1/vendor/bundle/jruby/3.1.0/gems/elasticsearch-transport-7.17.11/lib/elasticsearch/transport/transport/base.rb:288:in `perform_request'", "/Users/ritik.loomba/Documents/ELKStack/logstash-8.16.1/vendor/bundle/jruby/3.1.0/gems/elasticsearch-transport-7.17.11/lib/elasticsearch/transport/transport/http/manticore.rb:91:in `perform_request'", "/Users/ritik.loomba/Documents/ELKStack/logstash-8.16.1/vendor/bundle/jruby/3.1.0/gems/elasticsearch-transport-7.17.11/lib/elasticsearch/transport/client.rb:197:in `perform_request'", "/Users/ritik.loomba/Documents/ELKStack/logstash-8.16.1/vendor/bundle/jruby/3.1.0/gems/elasticsearch-7.17.11/lib/elasticsearch.rb:93:in `elasticsearch_validation_request'", "/Users/ritik.loomba/Documents/ELKStack/logstash-8.16.1/vendor/bundle/jruby/3.1.0/gems/elasticsearch-7.17.11/lib/elasticsearch.rb:51:in `verify_elasticsearch'", "/Users/ritik.loomba/Documents/ELKStack/logstash-8.16.1/vendor/bundle/jruby/3.1.0/gems/elasticsearch-7.17.11/lib/elasticsearch.rb:40:in `method_missing'", "/Users/ritik.loomba/Documents/ELKStack/logstash-8.16.1/vendor/bundle/jruby/3.1.0/gems/elasticsearch-api-7.17.11/lib/elasticsearch/api/actions/ping.rb:38:in `ping'", "/Users/ritik.loomba/Documents/ELKStack/logstash-8.16.1/vendor/bundle/jruby/3.1.0/gems/logstash-input-elasticsearch-4.20.4/lib/logstash/inputs/elasticsearch.rb:600:in `test_connection!'", "/Users/ritik.loomba/Documents/ELKStack/logstash-8.16.1/vendor/bundle/jruby/3.1.0/gems/logstash-input-elasticsearch-4.20.4/lib/logstash/inputs/elasticsearch.rb:326:in `register'", "/Users/ritik.loomba/Documents/ELKStack/logstash-8.16.1/vendor/bundle/jruby/3.1.0/gems/logstash-mixin-ecs_compatibility_support-1.3.0-java/lib/logstash/plugin_mixins/ecs_compatibility_support/target_check.rb:48:in `register'", "/Users/ritik.loomba/Documents/ELKStack/logstash-8.16.1/logstash-core/lib/logstash/java_pipeline.rb:246:in `block in register_plugins'", "org/jruby/RubyArray.java:1981:in `each'", "/Users/ritik.loomba/Documents/ELKStack/logstash-8.16.1/logstash-core/lib/logstash/java_pipeline.rb:245:in `register_plugins'", "/Users/ritik.loomba/Documents/ELKStack/logstash-8.16.1/logstash-core/lib/logstash/java_pipeline.rb:406:in `start_inputs'", "/Users/ritik.loomba/Documents/ELKStack/logstash-8.16.1/logstash-core/lib/logstash/java_pipeline.rb:330:in `start_workers'", "/Users/ritik.loomba/Documents/ELKStack/logstash-8.16.1/logstash-core/lib/logstash/java_pipeline.rb:199:in `run'", "/Users/ritik.loomba/Documents/ELKStack/logstash-8.16.1/logstash-core/lib/logstash/java_pipeline.rb:151:in `block in start'"], "pipeline.sources"=>["/Users/ritik.loomba/Documents/ELKStack/logstash-8.16.1/config/elasticsearchtofile.conf"], :thread=>"#<Thread:0x64eaad30 /Users/ritik.loomba/Documents/ELKStack/logstash-8.16.1/logstash-core/lib/logstash/java_pipeline.rb:139 run>"}
[2024-12-03T12:21:36,570][INFO ][logstash.javapipeline    ][main] Pipeline terminated {"pipeline.id"=>"main"}
[2024-12-03T12:21:36,577][ERROR][logstash.agent           ] Failed to execute action {:id=>:main, :action_type=>LogStash::ConvergeResult::FailedAction, :message=>"Could not execute action: PipelineAction::Create<main>, action_result: false", :backtrace=>nil}
[2024-12-03T12:21:36,588][INFO ][logstash.runner          ] Logstash shut down.
[2024-12-03T12:21:36,591][FATAL][org.logstash.Logstash    ] Logstash stopped processing because of an error: (SystemExit) exit
org.jruby.exceptions.SystemExit: (SystemExit) exit
	at org.jruby.RubyKernel.exit(org/jruby/RubyKernel.java:924) ~[jruby.jar:?]
	at org.jruby.RubyKernel.exit(org/jruby/RubyKernel.java:883) ~[jruby.jar:?]
	at Users.ritik_dot_loomba.Documents.ELKStack.logstash_minus_8_dot_16_dot_1.lib.bootstrap.environment.<main>(/Users/ritik.loomba/Documents/ELKStack/logstash-8.16.1/lib/bootstrap/environment.rb:90) ~[?:?]

You should either set ssl_verification_mode: none or provide a valid certificate configuration for the input

It worked without any error but file is not generated at the path mentioned in the output.

[2024-12-03T12:47:57,535][INFO ][logstash.javapipeline    ][main] Pipeline started {"pipeline.id"=>"main"}
[2024-12-03T12:47:57,536][INFO ][logstash.inputs.elasticsearch.searchafter][main][f8125e8606500ce8b146278c7ec5504b6799f66a70f1634ae757d60e8779aa73] Create point in time (PIT)
[2024-12-03T12:47:57,540][INFO ][logstash.inputs.elasticsearch.searchafter][main][f8125e8606500ce8b146278c7ec5504b6799f66a70f1634ae757d60e8779aa73] Query start
[2024-12-03T12:47:57,542][INFO ][logstash.agent           ] Pipelines running {:count=>1, :running_pipelines=>[:main], :non_running_pipelines=>[]}
[2024-12-03T12:47:57,547][INFO ][logstash.inputs.elasticsearch.searchafter][main][f8125e8606500ce8b146278c7ec5504b6799f66a70f1634ae757d60e8779aa73] Query completed
[2024-12-03T12:47:57,547][INFO ][logstash.inputs.elasticsearch.searchafter][main][f8125e8606500ce8b146278c7ec5504b6799f66a70f1634ae757d60e8779aa73] Closing point in time (PIT)
[2024-12-03T12:47:59,190][INFO ][logstash.javapipeline    ][main] Pipeline terminated {"pipeline.id"=>"main"}
[2024-12-03T12:47:59,567][INFO ][logstash.pipelinesregistry] Removed pipeline from registry successfully {:pipeline_id=>:main}
[2024-12-03T12:47:59,577][INFO ][logstash.runner          ] Logstash shut down.

You have not specified the index parameter for the input part of your pipeline. See: Elasticsearch input plugin | Logstash Reference [8.16] | Elastic

The default value is logstash-*, you'll need to set the index to match the name of the indices that store the data you're attempting to query.

index

  • Value type is string
  • Default value is "logstash-*"

The index or alias to search.

Thanks, Its working fine now.

I want to setup a logstash pipeline which will read data from Elasticsearch and load it to snowflake. It will be executed in two parts one is elasticsearch to s3 and then s3 to snowflake.

Does Creating a logstash pipeline which will unload data from elasticsearch to aws s3 in json format ,

  1. Can we capture records increamentally ?
  2. does that capture updates and deletes also, if yes how ?

Currently i am trying in local environment,

The documents pushed through the pipeline are the documents returned by your query.

Documents deleted will not show up in your query.

There are various ways to do incremental querying including things like querying a specific time range like only querying the last day and running the pipeline each day.

But in general the Elasticsearch input is not suited for incrementally "syncing" only changes or deletes

Thanks alot for your assistance. Then I think doing a full load is the best option.