Cannot get new connection from pool when same pipeline read and write to same index

[andsel]

It happens that when we have a pipeline composed of elasticsearch input that search all on an index and an elasticsearch output that deletes the documents just loaded from input we get this error:

Elasticsearch::Transport::Transport::Error: Cannot get new connection from pool.
  perform_request at /tmp/logstash-7.7.0/vendor/bundle/jruby/2.5.0/gems/elasticsearch-transport-5.0.5/lib/elasticsearch/transport/transport/base.rb:254
  perform_request at /tmp/logstash-7.7.0/vendor/bundle/jruby/2.5.0/gems/elasticsearch-transport-5.0.5/lib/elasticsearch/transport/transport/http/manticore.rb:67
  perform_request at /tmp/logstash-7.7.0/vendor/bundle/jruby/2.5.0/gems/elasticsearch-transport-5.0.5/lib/elasticsearch/transport/client.rb:131
           search at /tmp/logstash-7.7.0/vendor/bundle/jruby/2.5.0/gems/elasticsearch-api-5.0.5/lib/elasticsearch/api/actions/search.rb:183
   search_request at /tmp/logstash-7.7.0/vendor/bundle/jruby/2.5.0/gems/logstash-input-elasticsearch-4.6.0/lib/logstash/inputs/elasticsearch.rb:321
     do_run_slice at /tmp/logstash-7.7.0/vendor/bundle/jruby/2.5.0/gems/logstash-input-elasticsearch-4.6.0/lib/logstash/inputs/elasticsearch.rb:269
           do_run at /tmp/logstash-7.7.0/vendor/bundle/jruby/2.5.0/gems/logstash-input-elasticsearch-4.6.0/lib/logstash/inputs/elasticsearch.rb:257

• Version: 4.6.0
• Operating System: Linux
• Steps to Reproduce:
• run an elasticsearch node with docker

docker run -p 9200:9200 -p 9300:9300 -e "discovery.type=single-node" -e "ELASTIC_PASSWORD=changeme" docker.elastic.co/elasticsearch/elasticsearch:7.8.1

• fill 10^6 docs with data_filler_pipeline.conf later attached

bin/logstash -f data_filler_pipeline.conf

• use a pipeline similar to the one provided by the customer

bin/logstash -f sut_pipeline.conf

Pipeline used to fill data:

input {
  generator {
    message => "sample data"
    count => 1000000
  }
}

output {
  elasticsearch {
    index => "test_data"
    user => "elastic"
    password => "changeme"
  }
}

Pipeline used to test the problem:

input {
  elasticsearch {
    docinfo => true
    docinfo_fields => ["_id"]
    query => '{ "query": { "match_all": {} } }'
    scroll => "5m"
    slices => 2
    
    index => "test_data"
	user => "elastic"
    password => "changeme"
  }
}

output {
  elasticsearch {
	action => "delete"
	document_id => "%{[@metadata][_id]}"
	http_compression => true
	pool_max => 500
	pool_max_per_route => 80

	index => "test_data"
	user => "elastic"
    password => "changeme"
  }
}

[kares]

quick try turning on JRuby internal logging shows that no other Ruby exception is propagated before the pool error:
(suspected that some cause might be getting swallowed but if so definitely not one that originates in Ruby land)

[2020-07-28T12:59:51,219][INFO ][logstash.inputs.elasticsearch][main][e03d13a79d30e9bc0f9df932a089d2d52cab2943ead887f272793d64758de776] Slice starting {:slice_id=>0, :slices=>2}
[2020-07-28T12:59:51,228][INFO ][logstash.inputs.elasticsearch][main][e03d13a79d30e9bc0f9df932a089d2d52cab2943ead887f272793d64758de776] Slice starting {:slice_id=>1, :slices=>2}
[2020-07-28T12:59:51,331][INFO ][logstash.agent           ] Pipelines running {:count=>1, :running_pipelines=>[:main], :non_running_pipelines=>[]}
2020-07-28T12:59:51.546+02:00 [e03d13a79d30e9bc0f9df932a089d2d52cab2943ead887f272793d64758de776_slice_0] INFO TraceType : Exception raised: Elasticsearch::Transport::Transport::Error : Cannot get new connection from pool.
2020-07-28T12:59:51.547+02:00 [e03d13a79d30e9bc0f9df932a089d2d52cab2943ead887f272793d64758de776_slice_0] INFO TraceType : Backtrace generated:
  perform_request at /home/kares/workspace/work/elastic/logstash-7.7.1/vendor/bundle/jruby/2.5.0/gems/elasticsearch-transport-5.0.5/lib/elasticsearch/transport/transport/base.rb:254
  perform_request at /home/kares/workspace/work/elastic/logstash-7.7.1/vendor/bundle/jruby/2.5.0/gems/elasticsearch-transport-5.0.5/lib/elasticsearch/transport/transport/http/manticore.rb:67
  perform_request at /home/kares/workspace/work/elastic/logstash-7.7.1/vendor/bundle/jruby/2.5.0/gems/elasticsearch-transport-5.0.5/lib/elasticsearch/transport/client.rb:131
           search at /home/kares/workspace/work/elastic/logstash-7.7.1/vendor/bundle/jruby/2.5.0/gems/elasticsearch-api-5.0.5/lib/elasticsearch/api/actions/search.rb:183
   search_request at /home/kares/workspace/work/elastic/logstash-7.7.1/vendor/bundle/jruby/2.5.0/gems/logstash-input-elasticsearch-4.6.0/lib/logstash/inputs/elasticsearch.rb:321
     do_run_slice at /home/kares/workspace/work/elastic/logstash-7.7.1/vendor/bundle/jruby/2.5.0/gems/logstash-input-elasticsearch-4.6.0/lib/logstash/inputs/elasticsearch.rb:269
           do_run at /home/kares/workspace/work/elastic/logstash-7.7.1/vendor/bundle/jruby/2.5.0/gems/logstash-input-elasticsearch-4.6.0/lib/logstash/inputs/elasticsearch.rb:257

[andsel]

As @kares noted I missed to add the hosts and the test above with hosts works. However I was able to reproduce it. I moved from 10^6 docs to 10^7 to have the query running for more time and then artificially asked for a delete (just type DELETE on the LS console at sometime), the process is almost the same as the previous (but update the pipeline config files) and then when LS is working type DELETE on its console. I got this:

[2020-07-28T18:21:09,466][INFO ][logstash.agent           ] Successfully started Logstash API endpoint {:port=>9600}
DELETE
warning: thread "5d121bce64fece2224bfccf7d8f66f849c1069021dbce926aaf94c386df04690_slice_0" terminated with exception (report_on_exception is true):
Elasticsearch::Transport::Transport::Error: Cannot get new connection from pool.
      perform_request at /home/andrea/workspace/elastic_products/logstash-7.7.0/vendor/bundle/jruby/2.5.0/gems/elasticsearch-transport-5.0.5/lib/elasticsearch/transport/transport/base.rb:254
      perform_request at /home/andrea/workspace/elastic_products/logstash-7.7.0/vendor/bundle/jruby/2.5.0/gems/elasticsearch-transport-5.0.5/lib/elasticsearch/transport/transport/http/manticore.rb:67
      perform_request at /home/andrea/workspace/elastic_products/logstash-7.7.0/vendor/bundle/jruby/2.5.0/gems/elasticsearch-transport-5.0.5/lib/elasticsearch/transport/client.rb:131
               scroll at /home/andrea/workspace/elastic_products/logstash-7.7.0/vendor/bundle/jruby/2.5.0/gems/elasticsearch-api-5.0.5/lib/elasticsearch/api/actions/scroll.rb:61
       scroll_request at /home/andrea/workspace/elastic_products/logstash-7.7.0/vendor/bundle/jruby/2.5.0/gems/logstash-input-elasticsearch-4.6.0/lib/logstash/inputs/elasticsearch.rb:317
  process_next_scroll at /home/andrea/workspace/elastic_products/logstash-7.7.0/vendor/bundle/jruby/2.5.0/gems/logstash-input-elasticsearch-4.6.0/lib/logstash/inputs/elasticsearch.rb:285
         do_run_slice at /home/andrea/workspace/elastic_products/logstash-7.7.0/vendor/bundle/jruby/2.5.0/gems/logstash-input-elasticsearch-4.6.0/lib/logstash/inputs/elasticsearch.rb:277
               do_run at /home/andrea/workspace/elastic_products/logstash-7.7.0/vendor/bundle/jruby/2.5.0/gems/logstash-input-elasticsearch-4.6.0/lib/logstash/inputs/elasticsearch.rb:257

data_filler_pipeline.conf.txt
sut_pipeline.conf.txt

[yaauie]

It looks like we need to add better error handling for the managed threads when consuming multiple slices simultaneously. There is currently no error handling.

@andsel your latest test, in which you send a single delete for a non-existing document with the literal id %{[@metadata][id]} (since the DELETE you created manually will not have that field to replace), we aren't even deleting a document that is referenced by the scroll's search context.

Can we replicate without managed slices scrolling on the input?

[andsel]

Ok the DELETE is in the output and this make explode the input, that's running in a different thread. I'll try to replicate without the slices

[andsel]

I've removed the elasticsearch-output and the error happens less frequently. I suppose the problem is due to race condition in the ES client library, the logstash-input-elasticsearch when configured with slices create 1 thread per slice, and these threads concurrently call get_connection to the connection library using the same @client instance. Now the library to select a connection from the list of open connection has various policies, in particular the default is round robin which share state without protecting, and this could create missed updates to the the counter variable https://github.com/elastic/elasticsearch-ruby/blob/v5.0.5/elasticsearch-transport/lib/elasticsearch/transport/transport/connections/selector.rb#L51-L56.
This is still present in 6.8.2 https://github.com/elastic/elasticsearch-ruby/blob/v6.8.2/elasticsearch-transport/lib/elasticsearch/transport/transport/connections/selector.rb#L55-L59 but fixed starting from 7.2.0 https://github.com/elastic/elasticsearch-ruby/blob/v7.2.0/elasticsearch-transport/lib/elasticsearch/transport/transport/connections/selector.rb#L77

[yaauie]

@andsel good find, and 🤦 at a connection manager that wasn't threadsafe. I think we are blocked from upgrading the client library because we need backward-support for doc types, but maybe we can apply a patch to the dependency in our code?

[andsel]

Good idea, we could reopen the RoundRobin class in the plugin source and rewrite the select method in a thread safe manner