Add test for elasticsearch re-connection after network error & allow graceful shutdown

[belimawr]

Proposed commit message

When the Elasticsearch client fails to publish events, it ends up calling Close in the connection (that is reused). To cancel the in-flight requests, the context is cancelled and a new one is created to used in future requests.

The callback to check the version holds a reference to the connection via a closure, now the Elasticsearch client holds a pointer to that connection, so whenever Close is called, the callback can create a request with the new, not cancelled, context.

An integration test is added to ensure the
ES output can always recover from network errors.

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.

Disruptive User Impact

It's a bug fix, there is no disruptive user impact

## Author's Checklist

How to test this PR locally

Related issues

• Closes Elasticsearch output does not recover after connection failure #40705

## Use cases
## Screenshots
## Logs

[mergify]

This pull request is now in conflicts. Could you fix it? 🙏
To fixup this pull request, you can check out it locally. See documentation: https://help.github.com/articles/checking-out-pull-requests-locally/

git fetch upstream
git checkout -b fix-es-connection-issue upstream/fix-es-connection-issue
git merge upstream/main
git push upstream fix-es-connection-issue

[mergify]

This pull request does not have a backport label.
If this is a bug or security fix, could you label this PR @belimawr? 🙏.
For such, you'll need to label your PR with:

• The upcoming major version of the Elastic Stack
• The upcoming minor version of the Elastic Stack (if you're not pushing a breaking change)

To fixup this pull request, you need to add the backport labels for the needed
branches, such as:

• backport-8./d is the label to automatically backport to the 8./d branch. /d is the digit

[mergify]

backport-8.x has been added to help with the transition to the new branch 8.x.
If you don't need it please use backport-skip label and remove the backport-8.x label.

[elasticmachine]

Pinging @elastic/elastic-agent-data-plane (Team:Elastic-Agent-Data-Plane)

[AndersonQ]

LGTM, but I have a question. I'll approve once it's answered

[cmacknz]

Can we just find the places where we don't call connect and fix them? This code is only called within Beats and we can search for uses of esleg.

Also using context.Background() is a smell, the parent context should be an argument, which will have the compiler find all NewConnection uses for you so you can audit them to see if they use close inappropriately.

The connection here also don't really represent a network connection at the network level, it looks like it is a convenience wrapper around an HTTP client. From that perspective closing or having a connection level context doesn't make much sense, it is just a wrapper for closing idle connections.

Arguably the contexts should be set on a per request basis in all the places that call execRequest and then the cancellation should propagate through those individual calls.

[belimawr]

Can we just find the places where we don't call connect and fix them? This code is only called within Beats and we can search for uses of esleg.

I went with this approach to be on the safe side of not introducing the possibility of a panic for calling a nil function. The panic I got when running the tests seems to be coming form a Connection that is not used but it's closed. That is triggered by a test testing a failure scenario where the ES host is not reachable.

Anyways, I've been looking into that.

Arguably the contexts should be set on a per request basis in all the places that call execRequest and then the cancellation should propagate through those individual calls.

I'm not sure that would achieve the same effect we currently have. Currently reqsContext is used to cancel in-flight requests when the Connection needs to be shutdown, that is done by the Close method that is called by a different gorotine than the one waiting for the in-flight request(s) to finish.

The issues that are fixed by this new behaviour:

• Windows service for Beat does not stop when output is unreachable #40518
• Windows service for Beat does not stop gracefully #38666

An the PR fixing them:

• [libbeat] Stop publisher properly #40572

Honestly, I rather have this PR merged as is, so the issues listed above and #40705 are correctly fixed on main. Then we can create a follow up issue to refactor the code ensuring that:

• Methods creating requests accept a context
• In-flight requests can be cancelled when the Connection is closed by a different goroutine.

[belimawr]

I finally figured out the reason why Filebeat would panic when the connection was closed. It's an interesting corner case.
The root cause is if the publishing pipeline never tries to publish an event and Filebeat is shutdown, in that case the connection to ES was never used, however it is closed during the shutdown process, leading to a panic if cancelReqs is nil.

I could add some checks to ensure cancelReqs and reqsContext are not used if nil, however it feels cleaner to keep the previous behaviour of NewConnection returning a Connection that is safe to use without any change of behaviour on its methods.

[cmacknz]

That you have to account for future usage at all here feels wrong too. We need to fix this in a way that does not require knowing future uses of the code. Make it impossible to have this bug again.

Why do we even need a closure, was it just there because it was convenient? Can we just not have a closure anymore? It looks like if we saved the captured onConnect it could be a method on the client to avoid also capturing the connection.

beats/libbeat/outputs/elasticsearch/client.go

Lines 136 to 160 in 3c03c74

	conn.OnConnectCallback = func() error {
	globalCallbackRegistry.mutex.Lock()
	defer globalCallbackRegistry.mutex.Unlock()
	for _, callback := range globalCallbackRegistry.callbacks {
	err := callback(conn)
	if err != nil {
	return err
	}
	}
	if onConnect != nil {
	onConnect.mutex.Lock()
	defer onConnect.mutex.Unlock()
	for _, callback := range onConnect.callbacks {
	err := callback(conn)
	if err != nil {
	return err
	}
	}
	}
	return nil
	}

[cmacknz]

I want to make sure I follow the lifetime of the connection properly. The Connect and Publish calls come from here:

beats/libbeat/publisher/pipeline/client_worker.go

Lines 131 to 158 in 3c03c74

	// Try to (re)connect so we can publish batch
	if !connected {
	// Return batch to other output workers while we try to (re)connect
	batch.Cancelled()
	if reconnectAttempts == 0 {
	w.logger.Infof("Connecting to %v", w.client)
	} else {
	w.logger.Infof("Attempting to reconnect to %v with %d reconnect attempt(s)", w.client, reconnectAttempts)
	}
	err := w.client.Connect()
	connected = err == nil
	if connected {
	w.logger.Infof("Connection to %v established", w.client)
	reconnectAttempts = 0
	} else {
	w.logger.Errorf("Failed to connect to %v: %v", w.client, err)
	reconnectAttempts++
	}
	continue
	}
	if err := w.publishBatch(batch); err != nil {
	connected = false
	}
	}

The close comes from here:

beats/libbeat/outputs/backoff.go

Lines 60 to 64 in 3c03c74

	func (b *backoffClient) Publish(ctx context.Context, batch publisher.Batch) error {
	err := b.client.Publish(ctx, batch)
	if err != nil {
	b.client.Close()
	}

Can you make Connect accept a context.Context so that the context is actually tied to the lifetime of the connection the way it is supposed to be? Then the client worker run() function would be responsible for creating+cancelling the context. That you can't see the close actually happen in that loop since it is dependent on the client type implementation is also annoying but one thing at a time.

This would require touching the interface of every output but it looks like the correct place for the lifetime of the context to be managed.

It looks like we only have one other use of eslegclient for monitoring that is not a test.

beats/libbeat/monitoring/report/elasticsearch/elasticsearch.go

Lines 215 to 218 in 3c03c74

	for {
	// Select one configured endpoint by random and check if xpack is available
	client := r.out[rand.Intn(len(r.out))]
	err := client.Connect()

[marc-gr]

Can you make Connect accept a context.Context so that the context is actually tied to the lifetime of the connection the way it is supposed to be? Then the client worker run() function would be responsible for creating+cancelling the context. That you can't see the close actually happen in that loop since it is dependent on the client type implementation is also annoying but one thing at a time.

This would require touching the interface of every output but it looks like the correct place for the lifetime of the context to be managed.

👍 , and in addition if this is done it would require the rest of outputs to honor that new context, too. IIRC each uses different cancellation mechanisms atm.

[mergify]

This pull request is now in conflicts. Could you fix it? 🙏
To fixup this pull request, you can check out it locally. See documentation: https://help.github.com/articles/checking-out-pull-requests-locally/

git fetch upstream
git checkout -b fix-es-connection-issue upstream/fix-es-connection-issue
git merge upstream/main
git push upstream fix-es-connection-issue