Trouble using cipher with AWS SQS input plugin

[brian-maloney]

I'm using logstash 2.2.0 to read medium volume events from AWS SQS (running 64 input threads). We recently added the cipher plugin to address the fact that SQS offers no confirmed encryption at-rest capabilities. We had some issues with logstash crashing with stack traces like so:

NoMethodError: undefined method `padding=' for nil:NilClass
and
OpenSSL::Cipher::CipherError: key not specified

Turning on debugging I saw the following messages from the cipher plugin:

max_cipher_reuse[10] reached, total_cipher_uses = 11
max_cipher_reuse[10] reached, total_cipher_uses = 12

The only possible interpretation I could come to from this was that the cipher object was being called from multiple threads, despite the fact that this shouldn't be possible given the pipeline worker architecture per https://www.elastic.co/guide/en/logstash/2.2/pipeline.html

As a quick patch to get us up and running, I put a mutex around basically the entire code that uses the OpenSSL cipher object. I realize this probably isn't the right solution but I wanted to get this documented.

I'll create a pull request with my code for reference.

[bitsofinfo]

Interesting, yeah it was coded without any mutex because of that exact same assumption, only one filter instance per thread.... hmmm

The only possible interpretation I could come to from this was that the cipher object was being called from multiple threads, despite the fact that this shouldn't be possible given the pipeline worker architecture per https://www.elastic.co/guide/en/logstash/2.2/pipeline.html

[brian-maloney]

Oh, another thing, I'm using the default 8 pipeline workers. I haven't tried bumping that up to match the input workers, but the SynchronousQueue in between the inputs and pipeline workers really should make the number of threads on each side completely independent. I can't really explain why I was seeing that behavior but it should at least be pretty easy to replicate, just slam a bunch of events through SQS (free account works fine) with lots of input threads and it should crash.

[bitsofinfo]

Yeah, having a semaphore around that cipher sort of the defeats the whole purpose of re-using it across invocations (to avoid re-creating a new instance each invocation), especially if w/ this sqs input multiple threads are going through the same filter instance. Creates a bottleneck, next best solution would be to do away w/ reuse of cipher instance all together if the pipeline threading is not working as documented

[brian-maloney]

I was really concerned about performance as well since it's a bit of a bottleneck but the performance still seems pretty good, which I attribute to there still being multiple instances in the pipeline workers, just somehow they're getting called by multiple threads. I also have tested it with max_cipher_reuse => 1, which just makes the problem worse. Of course in a completely refactored filter function where the cipher object was initialized at the beginning of the function without regard to reuses, wouldn't it still be just as possible for that cipher object to be used more than once if the same filter object is reused (again, somehow, we don't know why this is happening) wouldn't it be possible for the cipher object to be damaged by a simultaneous call from another thread just as it as now. I'm definitely no Ruby expert (this is literally the first few lines of Ruby I've ever written) so I'm just guessing at what's happening.

I also tested this with logstash 2.3.0 since they reworked output threads just in case that had something to do with it, but the same issue persists, and the same patch fixes it.

[anandpathak]

I also had same problem when I used multiple kafka input plugin,( here https://discuss.elastic.co/t/logstash-fails-at-logstash-filter-cipher-plugin/63176)
@brian-maloney adding mutex solved the problem 👍

[bitsofinfo]

@anandpathak that link is a 404

[anandpathak]

Oh ! thank you @bitsofinfo for pointing it out, corrected !