[Filebeat] Fix reference leak in TCP and Unix socket inputs

[andrewkroh]

What does this PR do?

The tcp and unix input sources were leaking references causing a memory leak.
When an accepted connection ended inputsource/common.Closer was
supposed to delete the pointer that it held to the connection, but due to a code
error delete was being called on the wrong map.

Why is it important?

This fixes a memory leak with the tcp and unix inputs.

Checklist

• My code follows the style guidelines of this project
• I have commented my code, particularly in hard-to-understand areas
• I have made corresponding changes to the documentation
• I have made corresponding change to the default configuration files
• I have added tests that prove my fix is effective or that my feature works
• I have added an entry in CHANGELOG.next.asciidoc or CHANGELOG-developer.next.asciidoc.

[elasticmachine]

Pinging @elastic/integrations-services (Team:Services)

[urso]

Good find. Please backport to 7.8 as well before FF.

[elasticmachine]

💚 Build Succeeded

Expand to view the summary

Build stats

• Build Cause: [Pull request #19459 updated]

• Start Time: 2020-06-29T21:15:16.314+0000

• Duration: 25 min 5 sec

Test stats 🧪

Test	Results
Failed	0
Passed	543
Skipped	127
Total	670

[andrewkroh]

@urso Can you please take another look. My fix triggered race conditions so I took a different approach and dropped in a cancellable context in place of the common.Closer. d77ab50

[urso]

LGTM.

I consider context and WaitGroup quite low level primitives. They require you to 'track' usage and even free resources (cancel functon MUST be called). We've got some helpers here: https://pkg.go.dev/github.com/elastic/[email protected]/ctxtool?tab=doc, and some kind of with cancelation here: https://pkg.go.dev/github.com/elastic/go-concert?tab=doc#OnceSignaler

The context in Listener could be replaced with OnceSignaler and the stop could be replaced handled via 'WithFunc'.

For higher level primitives managing ownership of groups of go-routines check out the Group interface here: https://pkg.go.dev/github.com/elastic/[email protected]/unison?tab=doc#Group
We've got a sample implementation with TaskGroup. If you keep StopOnError false and do not return an error the implementation perfectly fits this use-case. starting the handler routine would look like this:

l.handlerGroup.Go(func(canceler unison.Canceler) error {
  ctx, cancel := ctxtool.WithFunc(ctxtool.FromCanceller(canceler), func() {
    conn.Close()
  })
  defer cancel()

  defer logp.Recover(...)

  ...

  return nil
})

The group combines shutdown signaling and the WaitGroup when calling 'Stop'. Once 'Stop' has been initiated any attempt to start another handler via Go will return an error, because we are about to shutdown.

[urso]

The handler go-routine is started always. Let's move cancellation setup local to the that go-routine. Consider to use ctxtool.WithFunc.

[andrewkroh]

Nice, I updated it to use ctxtool.WithFunc.

[ph]

fyi @vjsamuel this might affect you?

[vjsamuel]

Thanks. I will need to redo our internal input during the next rebase. Moving to context is the right thing to do however.