[Filebeat] Race condition fix in S3 input plugin

[darupedk]

During high load, Filebeat may crash due to the error channel c.errC having been closed before the last event has been acknowledged. This is caused by a missing critical section in done() which can result in c.refs no longer counting as expected.

[elasticmachine]

Since this is a community submitted pull request, a Jenkins build has not been kicked off automatically. Can an Elastic organization member please verify the contents of this patch and then kick off a build manually?

[elasticmachine]

Since this is a community submitted pull request, a Jenkins build has not been kicked off automatically. Can an Elastic organization member please verify the contents of this patch and then kick off a build manually?

[andrewkroh]

Fail() calls done() and holds the lock. That will result in a deadlock. I didn't look at the rest of the file, but wanted to call this out now. I can probably look at the rest of it tomorrow if any assistance is needed.

[darupedk]

I agree. That needs to be resolved.

Considering that we just want the latest error for now, we should be fine with the following change as only the counter needs to be protected against race conditions:

func (c *s3Context) Fail(err error) {
	// only care about the last error for now
	// TODO: add "Typed" error to error for context
	c.err = err
	c.done()
}

[urso]

if we have a race in 'done' due to concurrent calls messing with the reference counter, I wonder if Fail can be triggered concurrently as well. Maybe it would be better to have Failed like this (we do run tests with race detector enabled):

func (c *s3Context) Fail(err error) {
	// only care about the last error for now
	// TODO: add "Typed" error to error for context
	c.mux.Lock()
	c.err = err
	c.mux.Unlock()
	c.done()
}

or:

func (c *s3Context) Fail(err error) {
	c.setError(err)
	c.done()
}

func (c *s3Context) SetError(err error) {
	// only care about the last error for now
	// TODO: add "Typed" error to error for context
	c.mux.Lock()
	c.err = err
	c.mux.Unlock()
}

[darupedk]

I have been running with the fix provided in the latest commit for about 4 days now. The ingest pipeline is processing about 1.5 million log lines every 5 minutes and so far it has been stable. Without the fix, the software kept crashing after processing a few files from S3.

[kaiyan-sheng]

@darupedk Thanks for fixing and testing this!
@andrewkroh @urso Thanks for the review. I will do the backports.